Python SDK
phoson-sdkOfficial async Python client for the Phoson Core API. Fully typed, streaming-first, and supports Anthropic, OpenAI, and OpenRouter out of the box.
Installation
Install with uv (recommended) or pip.
uv add phoson-sdkpip install phoson-sdkQuick Start
Create a client, spin up an agent, start a conversation, and stream a response — in under 20 lines.
import asyncio
from phoson import PhosonClient
client = PhosonClient(api_key="pk_...", base_url="https://api.phoson.ai")
async def main():
# 1. Create an agent
agent = await client.agents.create(
name="My Assistant",
description="A helpful assistant",
memory_type="persistent",
)
# 2. Start a conversation
conv = await client.conversations.create(agent_id=agent.id, title="Hello")
# 3. Stream a response
async for event in client.messages.stream(conv.id, "Hello, how are you?"):
if event.type == "token":
print(event.content, end="", flush=True)
elif event.type == "run_done":
print(f"\n\nRun ID: {event.run_id}")
asyncio.run(main())Authentication
All requests are authenticated with an API key sent as a Bearer token. Get your key from the dashboard settings.
from phoson import PhosonClient
client = PhosonClient(api_key="pk_your_api_key_here")PhosonClient(api_key=os.environ["PHOSON_API_KEY"])PhosonClient
The top-level client that exposes all resource managers.
PhosonClient(*, api_key: str, base_url: str = "https://api.phoson.ai")api_key
str
Your Phoson API key (required).
base_url
str
API base URL. Defaults to https://api.phoson.ai.
| Parameter | Type | Description |
|---|---|---|
| api_key | str | Your Phoson API key (required). |
| base_url | str | API base URL. Defaults to https://api.phoson.ai. |
Agents
Agents are the core entities in Phoson. Each agent has its own memory, identity, and optional MCP tool integrations. Access via client.agents.
create
agent = await client.agents.create(
name="My Agent",
description="What this agent does",
memory_type="persistent", # "persistent" | "semantic", default "persistent"
)list
agents = await client.agents.list()get
agent = await client.agents.get(agent_id)update
agent = await client.agents.update(agent_id, name="New name", description="New desc")delete
await client.agents.delete(agent_id)attach_mcp / detach_mcp
Attach or detach an MCP server from an agent to give it tool-use capabilities.
await client.agents.attach_mcp(agent_id, mcp_id)
await client.agents.detach_mcp(agent_id, mcp_id)Conversations
A conversation is a thread of messages between a user and an agent. Access via client.conversations.
create
conv = await client.conversations.create(
agent_id=agent.id,
title="My conversation", # optional, default "New Conversation"
thread_id="custom-id", # optional
)list
convs = await client.conversations.list()
convs = await client.conversations.list(agent_id=agent.id) # filter by agentget
conv = await client.conversations.get(conversation_id)update
conv = await client.conversations.update(conversation_id, title="New title")delete
await client.conversations.delete(conversation_id)Messages
Send messages to a conversation and receive AI responses. Access via client.messages.
stream
Sends a message and returns an async iterator of StreamEvent objects. Recommended for interactive applications.
async for event in client.messages.stream(
conversation_id,
"Your message here",
model="claude-sonnet-4-6", # optional
provider="anthropic", # "anthropic" | "openai" | "openrouter"
temperature=0.7, # optional
max_tokens=4096, # optional
reasoning_effort="medium", # "low" | "medium" | "high", optional
):
match event.type:
case "token":
print(event.content, end="", flush=True)
case "reasoning":
print(f"[thinking] {event.content}")
case "tool_start":
print(f"→ Running tool: {event.tool}")
case "tool_end":
print(f"← Tool result: {event.result}")
case "run_done":
print(f"\nDone. Run ID: {event.run_id}")
case "error":
print(f"Error: {event.message}")send
Non-streaming version. Waits for the full response before returning.
message = await client.messages.send(
conversation_id,
"Your message here",
model="claude-sonnet-4-6",
provider="anthropic",
)
print(message.content)list
Retrieves the full message history of a conversation.
messages = await client.messages.list(conversation_id)
for msg in messages:
print(f"[{msg.role}] {msg.content}")Stream Events
Each event yielded by messages.stream() has a type field and type-specific fields.
token
content: str
reasoning
content: str
tool_start
tool: str, input: dict, tool_call_id: str
tool_end
tool: str, result: str, tool_call_id: str
run_done
content: str, run_id: str
error
message: str
| type | Fields |
|---|---|
| token | content: str |
| reasoning | content: str |
| tool_start | tool: str, input: dict, tool_call_id: str |
| tool_end | tool: str, result: str, tool_call_id: str |
| run_done | content: str, run_id: str |
| error | message: str |
Error Handling
The SDK raises PhosonAPIError for all API errors, exposing status_code and detail.
from phoson import PhosonAPIError
try:
agent = await client.agents.get("non-existent-id")
except PhosonAPIError as e:
print(e.status_code) # 404
print(e.detail) # "Agent not found"