Status: Actively Maintained - The Python SDK is actively maintained alongside the TypeScript SDK and provides full feature parity.
Chucky Client
The main client class for interacting with Chucky in Python.
Installation
Or with Poetry:
Import
from chucky import Chucky
Constructor
client = Chucky(
token: str,
url: str = 'wss://conjure.chucky.cloud/ws',
model: str = 'claude-sonnet-4-5-20250929',
system_prompt: Optional[str] = None,
max_turns: Optional[int] = None,
timeout: float = 30.0,
keepalive_interval: float = 60.0,
)
Parameters
| Parameter | Type | Required | Default | Description |
|---|
token | str | Yes | - | JWT authentication token |
url | str | No | 'wss://conjure.chucky.cloud/ws' | WebSocket server URL |
model | str | No | 'claude-sonnet-4-5-20250929' | Claude model to use |
system_prompt | str | No | None | System prompt |
max_turns | int | No | None | Max conversation turns |
timeout | float | No | 30.0 | Connection timeout (seconds) |
keepalive_interval | float | No | 60.0 | Keep-alive interval (seconds) |
Example
from chucky import Chucky
client = Chucky(
token='your-jwt-token',
model='claude-sonnet-4-5-20250929',
system_prompt='You are a helpful assistant.',
timeout=60.0,
)
Methods
prompt()
Send a one-shot prompt and get the response.
async def prompt(
message: str,
*,
model: Optional[str] = None,
system_prompt: Optional[str] = None,
tools: Optional[List[Tool]] = None,
max_turns: Optional[int] = None,
) -> PromptResult
Parameters
| Parameter | Type | Description |
|---|
message | str | The prompt message |
model | str | Override default model |
system_prompt | str | Override default system prompt |
tools | List[Tool] | Tools available to Claude |
max_turns | int | Max conversation turns |
Returns
PromptResult with the response.
Example
result = await client.prompt('What is the capital of France?')
print(result.result) # "The capital of France is Paris."
print(result.total_cost_usd) # 0.0012
stream()
Send a prompt with streaming response.
async def stream(
message: str,
*,
model: Optional[str] = None,
system_prompt: Optional[str] = None,
tools: Optional[List[Tool]] = None,
max_turns: Optional[int] = None,
) -> AsyncGenerator[StreamEvent, None]
Example
async for event in client.stream('Write a poem about coding'):
if event.type == 'message':
print(event.data, end='', flush=True)
elif event.type == 'tool_use':
print(f'\n[Using tool: {event.name}]')
create_session()
Create a new conversation session. Returns a Session immediately; connection happens on first send().
def create_session(
*,
model: Optional[str] = None,
system_prompt: Optional[str] = None,
tools: Optional[List[Tool]] = None,
max_turns: Optional[int] = None,
) -> Session
Example
# create_session is synchronous - returns Session immediately
session = client.create_session(
model='claude-sonnet-4-5-20250929',
system_prompt='You are a math tutor.',
)
# Connection happens on first send()
await session.send('What is calculus?')
async for msg in session.stream():
if msg.get('type') == 'result':
print(msg.get('result'))
await session.close()
close()
Close the client and release resources.
async def close() -> None
Example
Session Class
The Session class provides a multi-turn conversation interface.
Methods
send()
Send a message to the session. Use stream() to get the response.
async def send(message: str) -> None
stream()
Stream the response after sending a message.
async def stream() -> AsyncIterator[Dict[str, Any]]
type, message, etc.).
receive()
Alias for stream() for V2 SDK compatibility.
async def receive() -> AsyncIterator[Dict[str, Any]]
close()
Close the session and release resources.
async def close() -> None
Properties
session_id
The session’s unique identifier.
@property
def session_id(self) -> str
Session Example
session = client.create_session(model='claude-sonnet-4-5-20250929')
# First turn
await session.send('Hello, I am Alice')
async for msg in session.stream():
if msg.get('type') == 'assistant':
text = get_assistant_text(msg)
print(text)
if msg.get('type') == 'result':
break
# Second turn - maintains conversation context
await session.send('What is my name?')
async for msg in session.stream():
if msg.get('type') == 'result':
print(msg.get('result')) # "Your name is Alice."
await session.close()
Message Types
Messages are dictionaries matching the SDK protocol.
Result Message
{
'type': 'result',
'subtype': 'success', # or 'error'
'result': str, # Final response text
'session_id': str,
'total_cost_usd': float,
'duration_ms': int,
'num_turns': int,
}
Assistant Message
{
'type': 'assistant',
'message': {
'role': 'assistant',
'content': [
{'type': 'text', 'text': '...'},
{'type': 'tool_use', 'id': '...', 'name': '...', 'input': {...}},
]
}
}
Helper Functions
Use these helpers to extract text from messages:
from chucky import get_assistant_text, get_result_text
# Extract text from assistant message
text = get_assistant_text(msg) # Returns concatenated text blocks
# Extract result from result message
result = get_result_text(msg) # Returns result string or None
StreamEvent (Legacy)
For the legacy client.stream() API:
@dataclass
class StreamEvent:
type: str # 'message', 'result', 'error', 'done'
data: Any # The message dict
Complete Example
import asyncio
from chucky import (
Chucky,
create_token,
create_budget,
tool,
text_result,
get_assistant_text,
)
# Create a token (server-side only)
token = create_token(
user_id='user-123',
project_id='your-project-id', # From app.chucky.cloud
secret='your-hmac-secret', # From app.chucky.cloud
budget=create_budget(ai_dollars=1.00, compute_hours=1, window='day'),
)
# Define a tool
@tool('get_time', 'Get current time')
async def get_time() -> dict:
from datetime import datetime
return text_result(datetime.now().isoformat())
async def main():
# Create client
client = Chucky(
url='wss://conjure.chucky.cloud/ws',
token=token,
model='claude-sonnet-4-5-20250929',
)
try:
# Simple prompt
result = await client.prompt('What is 2 + 2?')
print(f"Answer: {result.get('result')}")
print(f"Cost: ${result.get('total_cost_usd', 0):.4f}")
# Multi-turn session with tools
session = client.create_session(tools=[get_time])
await session.send('What time is it?')
async for msg in session.stream():
if msg.get('type') == 'assistant':
print(get_assistant_text(msg))
if msg.get('type') == 'result':
print(f"Done: {msg.get('result')}")
break
await session.close()
finally:
await client.close()
asyncio.run(main())
Error Handling
Errors are returned as result messages with subtype: 'error':
from chucky import Chucky
client = Chucky(url='wss://conjure.chucky.cloud/ws', token='...')
try:
result = await client.prompt('Hello')
if result.get('subtype') == 'error':
print(f"Error: {result.get('result')}")
else:
print(result.get('result'))
except RuntimeError as e:
# Connection errors, timeouts
print(f'Connection error: {e}')
except Exception as e:
print(f'Unexpected error: {e}')
session = client.create_session()
await session.send('Hello')
async for msg in session.stream():
if msg.get('type') == 'error':
print(f"Error: {msg.get('error', {}).get('message')}")
break
