zai-adk-python

English | 简体中文

A Python SDK for building AI agents with LLM integration. Provides a unified interface for multiple LLM providers (OpenAI, Anthropic) with native tool calling, sandboxed execution, skills system, and observability.

Features

• Unified LLM Interface — Single API for OpenAI and Anthropic with provider-specific optimizations
• Tool System — FastAPI-style dependency injection with automatic JSON schema generation
• Sandboxed Execution — Secure filesystem and command isolation for agent tools
• Skills System — Reusable prompt instructions with hot-reload support
• MCP Integration — First-class support for Model Context Protocol servers
• Hook System — Pre-validation, post-review, and self-correction at 4 key points in the agent loop
• Event Streaming — Real-time agent events (thinking, tool calls, responses)
• Token Tracking — Built-in cost calculation and usage monitoring
• Observability — Laminar integration for tracing and debugging

Requirements

• Python >= 3.11
• uv for dependency management

Installation

# Clone the repository
git clone <repo-url>
cd zai-adk-python

# Install dependencies
uv sync --group dev

Configuration

Create a .env file in the project root:

# Anthropic
ANTHROPIC_BASE_URL=https://open.bigmodel.cn/api/anthropic
ANTHROPIC_API_KEY=your-api-key
ANTHROPIC_MODEL=glm-4.7

# OpenAI (optional)
OPENAI_BASE_URL=https://open.bigmodel.cn/api/paas/v4/
OPENAI_API_KEY=your-api-key
OPENAI_MODEL=glm-4.7

Quick Start

Basic Agent

import asyncio
from zai_adk import Agent
from zai_adk.llm.anthropic.chat import ChatAnthropic
from zai_adk.tools import tool

@tool("Calculate the sum of two numbers")
async def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

async def main():
    agent = Agent(
        llm=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
        tools=[add],
    )

    response = await agent.query("What is 25 + 17?")
    print(response)

asyncio.run(main())

Streaming Events

from zai_adk.agent import ToolCallEvent, ToolResultEvent, FinalResponseEvent

async for event in agent.query_stream("Help me with a task"):
    match event:
        case ToolCallEvent(tool=name, args=args):
            print(f"[Calling {name}]")
        case ToolResultEvent(tool=name, result=result):
            print(f"[Result from {name}]: {result}")
        case FinalResponseEvent(content=text):
            print(f"[Final]: {text}")

True Token Streaming

Enable real-time token-by-token streaming from the LLM with stream=True:

from zai_adk.agent import TextChunkEvent, ThinkingChunkEvent, FinalResponseEvent

async for event in agent.query_stream("Explain quantum computing", stream=True):
    match event:
        case ThinkingChunkEvent(content=delta):
            print(delta, end="", flush=True)   # thinking token delta
        case TextChunkEvent(content=delta):
            print(delta, end="", flush=True)   # text token delta
        case FinalResponseEvent(content=text):
            print(f"\nDone: {text}")

Key differences from default (stream=False):

	stream=False (default)	stream=True
First token latency	Waits for full response	Streams immediately
Event types	TextEvent (complete text)	TextChunkEvent per token + TextEvent (complete)
Thinking	ThinkingEvent (complete)	ThinkingChunkEvent per token + ThinkingEvent (complete)
Backward compat	—	Fully compatible, same tool execution flow

Enable extended thinking (GLM, DeepSeek):

from zai_adk.llm.openai.chat import ChatOpenAI

llm = ChatOpenAI(
    model="glm-5.1",
    thinking={"type": "enabled"},  # enable reasoning output
)
agent = Agent(llm=llm, tools=[])

async for event in agent.query_stream("Think carefully: 9.11 vs 9.8", stream=True):
    match event:
        case ThinkingChunkEvent(content=delta):
            print(f"[think] {delta}", end="", flush=True)
        case TextChunkEvent(content=delta):
            print(delta, end="", flush=True)

Sandboxed File Operations

from pathlib import Path
from zai_adk.sandbox import SandboxConfig
from zai_adk.tools.builtin.sandbox import bash, fs_read, fs_write

agent = Agent(
    llm=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
    tools=[bash, fs_read, fs_write],
    sandbox=SandboxConfig(
        kind="local",
        work_dir="./workspace",
        enforce_boundary=True,
    ),
)

Skills System

from zai_adk.skills import Skill, SkillsManager
from zai_adk.tools.builtin import skills

skills_manager = SkillsManager()
skills_manager.register(
    Skill(
        name="code-review",
        description="Review code for bugs and security issues",
        instructions="""You are a code reviewer...""",
    )
)

agent = Agent(
    llm=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
    tools=[skills],
    skills_manager=skills_manager,
)

Or use filesystem-based skills with auto-discovery:

agent = Agent(
    llm=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
    tools=[skills],
    skills_dir="./skills",  # Scans for SKILL.md files
)

Tool Creation

Use the @tool decorator with type hints for automatic schema generation:

from typing import Annotated
from zai_adk.tools import tool, Depends

# Simple tool
@tool("Get current weather")
async def get_weather(location: str, unit: str = "celsius") -> str:
    """Get the weather for a location."""
    return f"Weather in {location}: 22° {unit}"

# With dependency injection
@tool("Search the database")
async def search_db(
    query: str,
    db_conn: Annotated[DbConnection, Depends(get_db)],
) -> list[dict]:
    """Search the database for matching records."""
    return await db_conn.execute(query)

Dependency Injection

from zai_adk.sandbox import get_sandbox
from zai_adk.sandbox.base import Sandbox

@tool("Execute a shell command")
async def bash(
    command: str,
    sandbox: Annotated[Sandbox, Depends(get_sandbox)],
) -> str:
    """Run a command in the sandboxed environment."""
    result = await sandbox.exec(command)
    return result.stdout

Ephemeral Tools

Keep only the last N tool outputs in message history:

@tool("Read a file", ephemeral=1)
async def read_file(path: str) -> str:
    """Only the most recent result is kept in context."""
    return Path(path).read_text()

Agent Modes

CLI Mode (Default)

Agent stops automatically when the LLM returns text without tool calls:

agent = Agent(llm=..., tools=[...], mode="cli")
response = await agent.query("List all Python files")

Autonomous Mode

Agent requires an explicit done tool call to signal completion:

@tool("Signal task completion")
async def done(message: str) -> str:
    return f"TASK COMPLETE: {message}"

agent = Agent(llm=..., tools=[..., done], mode="autonomous")

Hook System

Inject validation, review, and self-correction logic at 4 key points in the agent loop — without modifying the loop itself.

from zai_adk import Agent, HookContext, HookResult, register_hook

# Block dangerous commands (global hook, applies to all agents)
def deny_dangerous(ctx: HookContext) -> HookResult:
    if ctx.tool_name == "bash":
        cmd = ctx.tool_args.get("command", "")
        if "rm -rf" in cmd:
            return HookResult(action="deny", reason="Blocked by safety policy")
    return HookResult(action="allow")

register_hook("pre_tool_use", deny_dangerous)

# Agent-level hook: enforce output format with self-correction
agent = Agent(llm=llm, tools=[...])

def require_summary(ctx: HookContext) -> HookResult:
    if "temperature" not in (ctx.response or "").lower():
        return HookResult(
            action="error",
            inject_message="Your response must include temperature data.",
        )
    return HookResult(action="allow")

agent.on("on_response", require_summary)

4 hook events cover the full agent cycle:

Event	Trigger Point	Use Cases
on_query_start	After user input, before LLM	Input validation, context injection
pre_tool_use	Before tool execution	Permission check, argument modification
post_tool_use	After tool execution	Output sanitization, logging
on_response	Before returning final response	Format enforcement, content review

See docs/hooks.md for the full technical documentation.

MCP Integration

from zai_adk.mcp import MCPServerConfig

agent = Agent(
    llm=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
    mcp_servers=[
        MCPServerConfig(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "./workspace"],
        ),
    ],
)

Protocol Support

The SDK supports multiple output protocols through a composable interface:

AG-UI Protocol

from zai_adk import Agent
from zai_adk.protocols.agui import AGUIProtocol

agent = Agent(llm=llm, tools=[tools])
agui = AGUIProtocol(agent)

async for event in agui.query_stream("Hello"):
    # Handle AG-UI events
    ...

SSE Streaming for HTTP

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from zai_adk import Agent
from zai_adk.protocols.agui import AGUIProtocol

app = FastAPI()
agent = Agent(llm=llm, tools=[tools])
agui = AGUIProtocol(agent)

@app.post("/chat")
async def chat(message: str):
    async def stream():
        async for sse in agui.query_stream_sse(message):
            yield sse

    return StreamingResponse(stream(), media_type="text/event-stream")

See docs/protocols.md for more details on implementing custom protocols.

Testing

# Run all tests
uv run pytest tests/ -v

# Skip tests requiring real LLM API calls
uv run pytest tests/ -v -m "not llm"

# Run specific test
uv run pytest tests/llm/test_anthropic_chat.py::test_anthropic_basic_chat -v

Examples

Example	Description
claude_code.py	Claude Code-style tools with sandboxed filesystem
hook_output_review.py	on_response hook: enforce output format with self-correction
hook_streaming.py	Streaming + hooks: real-time output review
hook_post_tool_use.py	post_tool_use hook: auto-mask secrets in tool output
skills_usage.py	Skills system patterns and hot-reload
mcp_usage.py	Model Context Protocol integration
todo_usage.py	Todo list management example

Run an example:

python -m examples.claude_code

Architecture

zai_adk/
├── agent/          # Agent loop with event streaming
├── hooks/          # Hook system — validation, review, self-correction
├── llm/            # LLM provider abstraction (OpenAI, Anthropic)
├── tools/          # Tool system with dependency injection
├── sandbox/        # Secure execution environments
├── skills/         # Reusable prompt instructions
├── mcp/            # Model Context Protocol integration
├── protocols/      # Protocol converter base interface
│   └── agui/       # AG-UI protocol implementation
├── tokens/         # Usage tracking and cost calculation
└── observability/  # Laminar tracing integration

License

MIT License