Open Agent SDK (Python)

Open-source Agent SDK that runs the full agent loop in-process — no subprocess or CLI required. Deploy anywhere: cloud, serverless, Docker, CI/CD.

Also available in TypeScript: open-agent-sdk-typescript and Go: open-agent-sdk-go

Get started

pip install open-agent-sdk

Set your API key:

export CODEANY_API_KEY=your-api-key

Third-party providers (e.g. OpenRouter) are supported via CODEANY_BASE_URL:

export CODEANY_BASE_URL=https://openrouter.ai/api
export CODEANY_API_KEY=sk-or-...
export CODEANY_MODEL=anthropic/claude-sonnet-4

Quick start

Streaming query

import asyncio
from open_agent_sdk import create_agent

async def main():
    agent = create_agent(
        allowed_tools=["Read", "Glob"],
        permission_mode="bypassPermissions",
    )
    async for event in agent.query("Read pyproject.toml and tell me the project name."):
        if event.get("type") == "assistant":
            for block in event["message"]["content"]:
                if "text" in block:
                    print(block["text"])

asyncio.run(main())

Simple blocking prompt

import asyncio
from open_agent_sdk import query

async def main():
    result = await query("What files are in this project?")
    print(result.text)
    print(f"Turns: {result.num_turns}, Tokens: {result.usage.input_tokens + result.usage.output_tokens}")

asyncio.run(main())

Multi-turn conversation

import asyncio
from open_agent_sdk import create_agent

async def main():
    agent = create_agent(max_turns=5)

    r1 = await agent.prompt('Create a file /tmp/hello.txt with "Hello World"')
    print(r1.text)

    r2 = await agent.prompt("Read back the file you just created")
    print(r2.text)

    print(f"Session messages: {len(agent.get_messages())}")

asyncio.run(main())

Custom tools

import asyncio
from open_agent_sdk import create_agent, get_all_base_tools
from open_agent_sdk.tools.types import define_tool

async def calc(input_data, ctx):
    expr = input_data["expression"]
    result = eval(expr)  # use ast.literal_eval in production
    return f"{expr} = {result}"

calculator = define_tool(
    name="Calculator",
    description="Evaluate a math expression",
    input_schema={
        "properties": {"expression": {"type": "string"}},
        "required": ["expression"],
    },
    is_read_only=True,
    call=calc,
)

async def main():
    agent = create_agent(tools=[*get_all_base_tools(), calculator])
    r = await agent.prompt("Calculate 2**10 * 3")
    print(r.text)

asyncio.run(main())

Custom tools (in-process MCP server)

import asyncio
from open_agent_sdk import create_agent, create_sdk_mcp_server
from open_agent_sdk.tools.types import define_tool

async def get_weather_handler(input_data, ctx):
    city = input_data["city"]
    return f"{city}: 22°C, sunny"

get_weather = define_tool(
    name="get_weather",
    description="Get the temperature for a city",
    input_schema={
        "properties": {"city": {"type": "string", "description": "City name"}},
        "required": ["city"],
    },
    is_read_only=True,
    call=get_weather_handler,
)

async def main():
    server = create_sdk_mcp_server(name="weather", tools=[get_weather])
    agent = create_agent(mcp_servers={"weather": server})
    async for msg in agent.query("What is the weather in Tokyo?"):
        if msg.get("type") == "result":
            print(f"Done: ${msg.get('total_cost_usd', 0):.4f}")

asyncio.run(main())

MCP server integration

import asyncio
from open_agent_sdk import create_agent

async def main():
    agent = create_agent(
        mcp_servers={
            "filesystem": {
                "type": "stdio",
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
            },
        },
    )

    result = await agent.prompt("List files in /tmp")
    print(result.text)

asyncio.run(main())

Subagents

import asyncio
from open_agent_sdk import create_agent

async def main():
    agent = create_agent(
        agents={
            "code-reviewer": {
                "description": "Expert code reviewer",
                "prompt": "Analyze code quality. Focus on security and performance.",
                "tools": ["Read", "Glob", "Grep"],
            },
        },
    )
    async for msg in agent.query("Use the code-reviewer agent to review src/main.py"):
        if msg.get("type") == "result":
            print("Done")

asyncio.run(main())

Permissions

import asyncio
from open_agent_sdk import create_agent

async def main():
    # Read-only agent — can only analyze, not modify
    agent = create_agent(
        allowed_tools=["Read", "Glob", "Grep"],
        permission_mode="dontAsk",
    )
    async for _ in agent.query("Review the code in src/ for best practices."):
        pass

asyncio.run(main())

API reference

Top-level functions

Function	Description
query(prompt, model=None, **options)	One-shot blocking query, returns QueryResult
create_agent(**options)	Create a reusable agent with session persistence
define_tool(name, description, input_schema, call)	Low-level tool definition helper
get_all_base_tools()	Get all 10 built-in tools
create_sdk_mcp_server(name, tools)	Bundle tools into an in-process MCP server
list_sessions()	List persisted sessions
fork_session(source_session_id)	Fork a session for branching

Agent methods

Method	Description
agent.query(prompt)	Streaming query, returns AsyncGenerator
agent.prompt(text)	Blocking query, returns QueryResult
agent.get_messages()	Get conversation history
agent.clear()	Reset session
agent.set_model(model)	Change model mid-session
agent.session_id	Get current session ID (property)

Options

Option	Type	Default	Description
model	str	claude-sonnet-4-6	LLM model ID
api_key	str	CODEANY_API_KEY	API key
base_url	str	—	Custom API endpoint
cwd	str	os.getcwd()	Working directory
system_prompt	str	—	System prompt override
append_system_prompt	str	—	Append to default system prompt
tools	list[ToolDefinition]	All built-in	Available tools
allowed_tools	list[str]	—	Tool allow-list
disallowed_tools	list[str]	—	Tool deny-list
permission_mode	str	bypassPermissions	default / acceptEdits / dontAsk / bypassPermissions / plan
can_use_tool	Callable	—	Custom permission callback
max_turns	int	10	Max agentic turns
max_budget_usd	float	—	Spending cap
max_tokens	int	16384	Max output tokens per turn
thinking	ThinkingConfig	disabled	Extended thinking
mcp_servers	dict	—	MCP server connections
agents	dict	—	Subagent definitions
resume	str	—	Resume session by ID
session_id	str	auto	Explicit session ID
persist_session	bool	False	Persist session to disk
env	dict[str, str]	—	Environment variables passed to agent

Environment variables

Variable	Description
CODEANY_API_KEY	API key (required)
CODEANY_AUTH_TOKEN	Alternative auth token
CODEANY_MODEL	Default model override
CODEANY_BASE_URL	Custom API endpoint

Built-in tools

Tool	Description
Bash	Execute shell commands
Read	Read files with line numbers
Write	Create / overwrite files
Edit	Precise string replacement in files
Glob	Find files by pattern
Grep	Search file contents with regex
WebFetch	Fetch and parse web content
WebSearch	Search the web
Agent	Spawn subagents for parallel work
AskUserQuestion	Ask the user for input

Architecture

┌──────────────────────────────────────────────────────┐
│                   Your Application                    │
│                                                       │
│   from open_agent_sdk import create_agent             │
└────────────────────────┬─────────────────────────────┘
                         │
              ┌──────────▼──────────┐
              │       Agent         │  Session state, tool pool,
              │  query() / prompt() │  MCP connections
              └──────────┬──────────┘
                         │
              ┌──────────▼──────────┐
              │    QueryEngine      │  Agentic loop:
              │   submit_message()  │  API call → tools → repeat
              └──────────┬──────────┘
                         │
         ┌───────────────┼───────────────┐
         │               │               │
   ┌─────▼─────┐  ┌─────▼─────┐  ┌─────▼─────┐
   │  LLM API  │  │  10 Tools │  │    MCP     │
   │  Client   │  │ Bash,Read │  │  Servers   │
   │ (streaming)│  │ Edit,...  │  │ stdio/SSE/ │
   └───────────┘  └───────────┘  │ HTTP/SDK   │
                                  └───────────┘

Key internals:

Component	Description
QueryEngine	Core agentic loop with auto-compact, retry, tool orchestration
Auto-compact	Summarizes conversation when context window fills up
Micro-compact	Truncates oversized tool results
Retry	Exponential backoff for rate limits and transient errors
Token estimation	Rough token counting for budget and compaction thresholds
Hook system	20 lifecycle events (PreToolUse, PostToolUse, SessionStart, …)
Session storage	Persist / resume / fork sessions on disk
Context injection	Git status + AGENT.md automatically injected into system prompt

Examples

#	File	Description
01	examples/01_simple_query.py	Streaming query with event loop
02	examples/02_multi_tool.py	Multi-tool orchestration
03	examples/03_multi_turn.py	Multi-turn session persistence
04	examples/04_prompt_api.py	Blocking prompt() API
05	examples/05_custom_system_prompt.py	Custom system prompt
06	examples/06_mcp_server.py	MCP server integration
07	examples/07_custom_tools.py	Custom tools with define_tool()
08	examples/08_query_function.py	query() function API
09	examples/09_subagents.py	Subagent delegation
10	examples/10_permissions.py	Read-only agent with tool restrictions
11	examples/11_custom_mcp_tools.py	define_tool() + create_sdk_mcp_server()

Run any example:

python examples/01_simple_query.py

License

MIT