Recipes.
Six runnable examples, from a CLI chatbot to durable stream recovery.
Recipe 1: CLI chatbot
A complete interactive CLI chatbot with device-flow auth and streaming.
import asyncio
from pathlib import Path
from uuid import uuid4
from matilda_client import MatildaClient, create_file_token_store
async def main():
token_store = create_file_token_store(Path.home() / ".matilda" / "tokens.json")
async with MatildaClient() as client:
# Authenticate if needed
if not await client.auth.get_tokens():
print("Starting device flow authentication...")
await client.auth.login_with_device_flow(
client_id="matilda-code",
token_store=token_store.store,
token_lock=token_store.lock,
)
print("Authenticated!")
conversation_id = str(uuid4())
while True:
user_input = await asyncio.to_thread(input, "\nYou: ")
if not user_input.strip() or user_input.lower() == "exit":
break
print("Matilda: ", end="", flush=True)
async for chunk in client.chat.stream_text(
input=user_input, conversation_id=conversation_id
):
print(chunk, end="", flush=True)
print()
asyncio.run(main())Recipe 2: File Q&A
Upload a document and ask questions about it.
import asyncio
from uuid import uuid4
from matilda_client import MatildaClient
async def main():
async with MatildaClient() as client:
# Authenticate with device flow — token refresh is handled automatically
if not await client.auth.get_tokens():
await client.auth.login_with_device_flow(client_id="matilda-code")
# Upload a file
upload = await client.files.upload(
"report.pdf",
content_type="application/pdf",
on_progress=lambda pct: print(f"\rUploading: {pct}%", end="", flush=True),
)
print(f"\nUploaded: {upload.file_id} ({upload.status})")
# A conversation_id is required for follow-ups to share history — omitting it
# auto-creates a new conversation per call.
conversation_id = str(uuid4())
# Ask a question about it
response = await client.chat.create(
input="Summarise the key findings in this report.",
file_ids=[upload.file_id],
conversation_id=conversation_id,
)
print(response.output_text)
# Follow-up question in the same conversation
follow_up = await client.chat.create(
input="What are the recommendations?",
file_ids=[upload.file_id],
conversation_id=conversation_id,
)
print(follow_up.output_text)
asyncio.run(main())Recipe 3: Conversation history browser
List, paginate, and inspect conversation history.
import asyncio
from matilda_client import MatildaClient
async def main():
async with MatildaClient() as client:
# Authenticate with device flow — token refresh is handled automatically
if not await client.auth.get_tokens():
await client.auth.login_with_device_flow(client_id="matilda-code")
# List first page
offset = 0
limit = 10
page = await client.conversations.list(limit=limit, offset=offset)
print(f"Total conversations: {page['total']}\n")
for conv in page["conversations"]:
print(f"[{conv['id']}] {conv['title']}")
print(f" Updated: {conv['updatedAt']}\n")
# Load next page
offset += limit
if offset < page["total"]:
page = await client.conversations.list(limit=limit, offset=offset)
for conv in page["conversations"]:
print(f"[{conv['id']}] {conv['title']}")
# Retrieve a full conversation
if page["conversations"]:
full = await client.conversations.retrieve(page["conversations"][0]["id"])
print(f"\n--- {full['title']} ---")
for msg in full["messages"]:
print(f"\n[{msg['role'].upper()}]")
print(msg["content"])
if msg.get("feedback"):
print(f" Feedback: {msg['feedback']}")
asyncio.run(main())Recipe 4: Durable stream recovery
Start a stream, simulate a disconnect, and resume from the last cursor.
import asyncio
from uuid import uuid4
from matilda_client import MatildaClient
async def main():
async with MatildaClient() as client:
# Authenticate with device flow — token refresh is handled automatically
if not await client.auth.get_tokens():
await client.auth.login_with_device_flow(client_id="matilda-code")
conversation_id = str(uuid4())
stream_id = None
last_event_id = None
received_text = ""
# Start streaming — simulate disconnect after a few events
print("Starting stream...")
try:
async for event in client.chat.stream(
input="Write a very long, detailed essay about the history of computing.",
conversation_id=conversation_id,
):
if event.type == "response.created":
stream_id = event.stream_id
if event.type == "response.cursor":
last_event_id = event.last_event_id
if event.type == "response.output_text.delta":
received_text += event.delta
# Simulate disconnect after 500 chars
if len(received_text) > 500:
print("\n--- Simulated disconnect ---")
break
if event.type == "response.completed":
print("Stream completed naturally.")
except Exception as err:
print(f"Disconnected: {err}")
print(f"Received {len(received_text)} chars before disconnect.")
# Check if the stream is still active
if stream_id:
active = await client.chat.active_stream(conversation_id)
print(f"Stream status: {active['status']}")
if active["status"] == "active":
print("\n--- Resuming ---")
async for event in client.chat.resume(
stream_id, last_event_id=last_event_id
):
if event.type == "response.output_text.delta":
received_text += event.delta
print(event.delta, end="", flush=True)
if event.type == "response.completed":
print("\n\n--- Resume complete ---")
print(f"Total received: {len(received_text)} chars")
asyncio.run(main())Recipe 5: Multi-environment setup
Run staging and production clients in the same process. Each instance manages its own auth independently.
import asyncio
from matilda_client import MatildaClient
async def main():
async with (
MatildaClient(base_url="https://staging.matilda.maincode.com/api") as staging,
MatildaClient(base_url="https://matilda.maincode.com/api") as production,
):
# Authenticate each instance independently — device flow or browser login
if not await staging.auth.get_tokens():
await staging.auth.login_with_device_flow(client_id="matilda-code")
if not await production.auth.get_tokens():
await production.auth.login_with_browser(client_id="matilda-code")
# Run the same prompt against both environments
staging_response, prod_response = await asyncio.gather(
staging.chat.create_text(input="Explain quantum entanglement."),
production.chat.create_text(input="Explain quantum entanglement."),
)
print("Staging:", staging_response)
print("Production:", prod_response)
# Instances are fully isolated — each manages its own token lifecycle,
# and config is immutable after construction, so one instance's setup
# can never leak into another.
asyncio.run(main())Recipe 6: Custom token store
Implement StorageAdapter to store tokens in a database or other custom backend.
import asyncio
from matilda_client import MatildaClient
# Example: a database-backed token store
class DatabaseTokenStore:
def __init__(self, db):
self._db = db
async def get(self, key: str) -> str | None:
row = await self._db.fetchone("SELECT value FROM tokens WHERE key = ?", (key,))
return row["value"] if row else None
async def set(self, key: str, value: str) -> None:
await self._db.execute(
"INSERT INTO tokens (key, value) VALUES (?, ?) "
"ON CONFLICT (key) DO UPDATE SET value = excluded.value",
(key, value),
)
async def remove(self, key: str) -> None:
await self._db.execute("DELETE FROM tokens WHERE key = ?", (key,))
async def main():
token_store = DatabaseTokenStore(my_database)
async with MatildaClient() as client:
await client.auth.login_with_device_flow(
client_id="matilda-code",
token_store=token_store,
# No cross-process lock needed — the database handles concurrency
)
# Tokens are now persisted in the database and survive process restarts
response = await client.chat.create(input="Hello!")
print(response.output_text)
asyncio.run(main())