Python SDK

phoson-sdk

Official async Python client for the Phoson Core API. Fully typed, streaming-first, and supports Anthropic, OpenAI, and OpenRouter out of the box.

$uv add phoson-sdk
Python 3.11+uv recommendedAsync nativeFully typedpip compatible

Installation

Install with uv (recommended) or pip.

bash
uv add phoson-sdk
bash
pip install phoson-sdk

Quick Start

Create a client, spin up an agent, start a conversation, and stream a response — in under 20 lines.

main.py
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.

python
from phoson import PhosonClient

client = PhosonClient(api_key="pk_your_api_key_here")
Security: Never hardcode API keys in source code. Use environment variables: PhosonClient(api_key=os.environ["PHOSON_API_KEY"])

PhosonClient

The top-level client that exposes all resource managers.

python
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.

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

python
agent = await client.agents.create(
    name="My Agent",
    description="What this agent does",
    memory_type="persistent",  # "persistent" | "semantic", default "persistent"
)

list

python
agents = await client.agents.list()

get

python
agent = await client.agents.get(agent_id)

update

python
agent = await client.agents.update(agent_id, name="New name", description="New desc")

delete

python
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.

python
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

python
conv = await client.conversations.create(
    agent_id=agent.id,
    title="My conversation",   # optional, default "New Conversation"
    thread_id="custom-id",     # optional
)

list

python
convs = await client.conversations.list()
convs = await client.conversations.list(agent_id=agent.id)  # filter by agent

get

python
conv = await client.conversations.get(conversation_id)

update

python
conv = await client.conversations.update(conversation_id, title="New title")

delete

python
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.

stream.py
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.

python
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.

python
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

Error Handling

The SDK raises PhosonAPIError for all API errors, exposing status_code and detail.

python
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"