Back to Details

agentica-sdk

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Agentica SDK Reference (v0.3.1)

Agentica SDK 参考文档（v0.3.1）

Build AI agents in Python using the Agentica framework. Agents can implement functions, maintain state, use tools, and coordinate with each other.
使用Agentica框架在Python中构建AI Agent。Agent可以实现函数、维护状态、使用工具并相互协作。

When to Use

适用场景

Use this skill when:
  • Building new Python agents
  • Adding agentic capabilities to existing code
  • Integrating MCP tools with agents
  • Implementing multi-agent orchestration
  • Debugging agent behavior
在以下场景使用本技能：
  • 构建新的Python Agent
  • 为现有代码添加Agent能力
  • 将MCP工具与Agent集成
  • 实现多Agent编排
  • 调试Agent行为

Quick Start

快速开始

Agentic Function (simplest)

基础Agentic函数

python
from agentica import agentic

@agentic()
async def add(a: int, b: int) -> int:
    """Returns the sum of a and b"""
    ...

result = await add(1, 2)  # Agent computes: 3
python
from agentica import agentic

@agentic()
async def add(a: int, b: int) -> int:
    """Returns the sum of a and b"""
    ...

result = await add(1, 2)  # Agent computes: 3

Spawned Agent (more control)

可控制的Spawned Agent

python
from agentica import spawn

agent = await spawn(premise="You are a truth-teller.")
result: bool = await agent.call(bool, "The Earth is flat")
python
from agentica import spawn

agent = await spawn(premise="You are a truth-teller.")
result: bool = await agent.call(bool, "The Earth is flat")

Returns: False

Returns: False

undefined
undefined

Core Patterns

核心模式

Return Types

返回类型

python
undefined
python
undefined

String (default)

String (default)

result = await agent.call("What is 2+2?")
result = await agent.call("What is 2+2?")

Typed output

Typed output

result: int = await agent.call(int, "What is 2+2?") result: dict[str, int] = await agent.call(dict[str, int], "Count items")
result: int = await agent.call(int, "What is 2+2?") result: dict[str, int] = await agent.call(dict[str, int], "Count items")

Side-effects only

Side-effects only

await agent.call(None, "Send message to John")
undefined
await agent.call(None, "Send message to John")
undefined

Premise vs System Prompt

Premise与系统提示词

python
undefined
python
undefined

Premise: adds to default system prompt

Premise: 追加到默认系统提示词

agent = await spawn(premise="You are a math expert.")
agent = await spawn(premise="You are a math expert.")

System: full control (replaces default)

System: 完全控制（替换默认提示词）

agent = await spawn(system="You are a JSON-only responder.")
undefined
agent = await spawn(system="You are a JSON-only responder.")
undefined

Passing Tools (Scope)

传递工具（Scope）

python
from agentica import agentic, spawn
python
from agentica import agentic, spawn

In decorator

在装饰器中指定

@agentic(scope={'web_search': web_search_fn}) async def researcher(query: str) -> str: """Research a topic.""" ...
@agentic(scope={'web_search': web_search_fn}) async def researcher(query: str) -> str: """Research a topic.""" ...

In spawn

在spawn中指定

agent = await spawn( premise="Data analyzer", scope={"analyze": custom_analyzer} )
agent = await spawn( premise="Data analyzer", scope={"analyze": custom_analyzer} )

Per-call scope

单次调用时指定Scope

result = await agent.call( dict[str, int], "Analyze the dataset", dataset=data, # Available as 'dataset' analyzer=custom_fn # Available as 'analyzer' )
undefined
result = await agent.call( dict[str, int], "Analyze the dataset", dataset=data, # 可通过'dataset'访问 analyzer=custom_fn # 可通过'analyzer'访问 )
undefined

SDK Integration Pattern

SDK集成模式

python
from slack_sdk import WebClient

slack = WebClient(token=SLACK_TOKEN)
python
from slack_sdk import WebClient

slack = WebClient(token=SLACK_TOKEN)

Extract specific methods

提取特定方法

@agentic(scope={ 'list_users': slack.users_list, 'send_message': slack.chat_postMessage }) async def team_notifier(message: str) -> None: """Send team notifications.""" ...
undefined
@agentic(scope={ 'list_users': slack.users_list, 'send_message': slack.chat_postMessage }) async def team_notifier(message: str) -> None: """Send team notifications.""" ...
undefined

Agent Instantiation

Agent实例化

spawn() - Async (most cases)

spawn() - 异步方式（大多数场景）

python
agent = await spawn(premise="Helpful assistant")
python
agent = await spawn(premise="Helpful assistant")

Agent() - Sync (for 
__init__
)

Agent() - 同步方式（用于
__init__
）

python
from agentica.agent import Agent

class CustomAgent:
    def __init__(self):
        # Synchronous - use Agent() not spawn()
        self._brain = Agent(
            premise="Specialized assistant",
            scope={"tool": some_tool}
        )

    async def run(self, task: str) -> str:
        return await self._brain(str, task)
python
from agentica.agent import Agent

class CustomAgent:
    def __init__(self):
        # 同步场景 - 使用Agent()而非spawn()
        self._brain = Agent(
            premise="Specialized assistant",
            scope={"tool": some_tool}
        )

    async def run(self, task: str) -> str:
        return await self._brain(str, task)

Model Selection

模型选择

python
undefined
python
undefined

In spawn

在spawn中指定

agent = await spawn( premise="Fast responses", model="openai:gpt-5" # Default: openai:gpt-4.1 )
agent = await spawn( premise="Fast responses", model="openai:gpt-5" # 默认值: openai:gpt-4.1 )

In decorator

在装饰器中指定

@agentic(model="anthropic:claude-sonnet-4.5") async def analyze(text: str) -> dict: """Analyze text.""" ...

**Available models:**
- `openai:gpt-3.5-turbo`, `openai:gpt-4o`, `openai:gpt-4.1`, `openai:gpt-5`
- `anthropic:claude-sonnet-4`, `anthropic:claude-opus-4.1`
- `anthropic:claude-sonnet-4.5`, `anthropic:claude-opus-4.5`
- Any OpenRouter slug (e.g., `google/gemini-2.5-flash`)
@agentic(model="anthropic:claude-sonnet-4.5") async def analyze(text: str) -> dict: """Analyze text.""" ...

**支持的模型：**
- `openai:gpt-3.5-turbo`, `openai:gpt-4o`, `openai:gpt-4.1`, `openai:gpt-5`
- `anthropic:claude-sonnet-4`, `anthropic:claude-opus-4.1`
- `anthropic:claude-sonnet-4.5`, `anthropic:claude-opus-4.5`
- 任何OpenRouter标识（例如：`google/gemini-2.5-flash`）

Persistence (Stateful Agents)

Agent持久化（有状态Agent）

python
@agentic(persist=True)
async def chatbot(message: str) -> str:
    """Remembers conversation history."""
    ...

await chatbot("My name is Alice")
await chatbot("What's my name?")  # Knows: Alice
For 
spawn()
agents, state is automatic across calls to the same instance.
python
@agentic(persist=True)
async def chatbot(message: str) -> str:
    """Remembers conversation history."""
    ...

await chatbot("My name is Alice")
await chatbot("What's my name?")  # 会返回: Alice
对于通过
spawn()
创建的Agent，同一实例的多次调用会自动保留状态。

Token Limits

Token限制

python
from agentica import spawn, MaxTokens
python
from agentica import spawn, MaxTokens

Simple limit

简单限制

agent = await spawn( premise="Brief responses", max_tokens=500 )
agent = await spawn( premise="Brief responses", max_tokens=500 )

Fine-grained control

细粒度控制

agent = await spawn( premise="Controlled output", max_tokens=MaxTokens( per_invocation=5000, # Total across all rounds per_round=1000, # Per inference round rounds=5 # Max inference rounds ) )
undefined
agent = await spawn( premise="Controlled output", max_tokens=MaxTokens( per_invocation=5000, # 所有轮次的总Token限制 per_round=1000, # 单轮推理的Token限制 rounds=5 # 最大推理轮次 ) )
undefined

Token Usage Tracking

Token使用量追踪

python
from agentica import spawn, last_usage, total_usage

agent = await spawn(premise="You are helpful.")
await agent.call(str, "Hello!")
python
from agentica import spawn, last_usage, total_usage

agent = await spawn(premise="You are helpful.")
await agent.call(str, "Hello!")

Agent method

Agent方法

usage = agent.last_usage() print(f"Last: {usage.input_tokens} in, {usage.output_tokens} out")
usage = agent.total_usage() print(f"Total: {usage.total_tokens} processed")
usage = agent.last_usage() print(f"Last: {usage.input_tokens} in, {usage.output_tokens} out")
usage = agent.total_usage() print(f"Total: {usage.total_tokens} processed")

For @agentic functions

对于@agentic装饰的函数

@agentic() async def my_fn(x: str) -> str: ...
await my_fn("test") print(last_usage(my_fn)) print(total_usage(my_fn))
undefined
@agentic() async def my_fn(x: str) -> str: ...
await my_fn("test") print(last_usage(my_fn)) print(total_usage(my_fn))
undefined

Streaming

流式输出

python
from agentica import spawn
from agentica.logging.loggers import StreamLogger
import asyncio

agent = await spawn(premise="You are helpful.")

stream = StreamLogger()
with stream:
    result = asyncio.create_task(
        agent.call(bool, "Is Paris the capital of France?")
    )
python
from agentica import spawn
from agentica.logging.loggers import StreamLogger
import asyncio

agent = await spawn(premise="You are helpful.")

stream = StreamLogger()
with stream:
    result = asyncio.create_task(
        agent.call(bool, "Is Paris the capital of France?")
    )

Consume stream FIRST for live output

先消费流以获取实时输出

async for chunk in stream: print(chunk.content, end="", flush=True)
async for chunk in stream: print(chunk.content, end="", flush=True)

chunk.role is 'user', 'agent', or 'system'

chunk.role 可选值为 'user', 'agent' 或 'system'

Then await result

然后等待结果完成

final = await result
undefined
final = await result
undefined

MCP Integration

MCP集成

python
from agentica import spawn, agentic
python
from agentica import spawn, agentic

Via config file

通过配置文件

agent = await spawn( premise="Tool-using agent", mcp="path/to/mcp_config.json" )
@agentic(mcp="path/to/mcp_config.json") async def tool_user(query: str) -> str: """Uses MCP tools.""" ...

**mcp_config.json format:**
```json
{
  "mcpServers": {
    "tavily-remote-mcp": {
      "command": "npx -y mcp-remote https://mcp.tavily.com/mcp/?tavilyApiKey=<key>",
      "env": {}
    }
  }
}
agent = await spawn( premise="Tool-using agent", mcp="path/to/mcp_config.json" )
@agentic(mcp="path/to/mcp_config.json") async def tool_user(query: str) -> str: """Uses MCP tools.""" ...

**mcp_config.json格式：**
```json
{
  "mcpServers": {
    "tavily-remote-mcp": {
      "command": "npx -y mcp-remote https://mcp.tavily.com/mcp/?tavilyApiKey=<key>",
      "env": {}
    }
  }
}

Logging

日志

Default Behavior

默认行为

  • Prints to stdout with colors
  • Writes to 
    ./logs/agent-<id>.log
  • 带颜色打印到标准输出
  • 写入到
    ./logs/agent-<id>.log
    文件

Contextual Logging

上下文日志

python
from agentica.logging.loggers import FileLogger, PrintLogger
from agentica.logging.agent_logger import NoLogging
python
from agentica.logging.loggers import FileLogger, PrintLogger
from agentica.logging.agent_logger import NoLogging

File only

仅输出到文件

with FileLogger(): agent = await spawn(premise="Debug agent") await agent.call(int, "Calculate")
with FileLogger(): agent = await spawn(premise="Debug agent") await agent.call(int, "Calculate")

Silent

静默模式

with NoLogging(): agent = await spawn(premise="Silent agent")
undefined
with NoLogging(): agent = await spawn(premise="Silent agent")
undefined

Per-Agent Logging

单Agent日志配置

python
undefined
python
undefined

Listeners are in agent_listener submodule (NOT exported from agentica.logging)

监听器位于agent_listener子模块（并非从agentica.logging导出）

from agentica.logging.agent_listener import ( PrintOnlyListener, # Console output only FileOnlyListener, # File logging only StandardListener, # Both console + file (default) NoopListener, # Silent - no logging )
agent = await spawn( premise="Custom logging", listener=PrintOnlyListener )
from agentica.logging.agent_listener import ( PrintOnlyListener, # 仅控制台输出 FileOnlyListener, # 仅文件日志 StandardListener, # 同时输出到控制台和文件（默认） NoopListener, # 静默模式 - 无日志 )
agent = await spawn( premise="Custom logging", listener=PrintOnlyListener )

Silent agent

静默Agent

agent = await spawn( premise="Silent agent", listener=NoopListener )
undefined
agent = await spawn( premise="Silent agent", listener=NoopListener )
undefined

Global Config

全局配置

python
from agentica.logging.agent_listener import (
    set_default_agent_listener,
    get_default_agent_listener,
    PrintOnlyListener,
)

set_default_agent_listener(PrintOnlyListener)
set_default_agent_listener(None)  # Disable all
python
from agentica.logging.agent_listener import (
    set_default_agent_listener,
    get_default_agent_listener,
    PrintOnlyListener,
)

set_default_agent_listener(PrintOnlyListener)
set_default_agent_listener(None)  # 禁用所有日志

Error Handling

错误处理

python
from agentica.errors import (
    AgenticaError,           # Base for all SDK errors
    RateLimitError,          # Rate limiting
    InferenceError,          # HTTP errors from inference
    MaxTokensError,          # Token limit exceeded
    MaxRoundsError,          # Max inference rounds exceeded
    ContentFilteringError,   # Content filtered
    APIConnectionError,      # Network issues
    APITimeoutError,         # Request timeout
    InsufficientCreditsError,# Out of credits
    OverloadedError,         # Server overloaded
    ServerError,             # Generic server error
)

try:
    result = await agent.call(str, "Do something")
except RateLimitError:
    await asyncio.sleep(60)
    result = await agent.call(str, "Do something")
except MaxTokensError:
    # Reduce scope or increase limits
    pass
except ContentFilteringError:
    # Content was filtered
    pass
except InferenceError as e:
    logger.error(f"Inference failed: {e}")
except AgenticaError as e:
    logger.error(f"SDK error: {e}")
python
from agentica.errors import (
    AgenticaError,           # 所有SDK错误的基类
    RateLimitError,          # 速率限制
    InferenceError,          # 推理服务HTTP错误
    MaxTokensError,          # 超出Token限制
    MaxRoundsError,          # 超出最大推理轮次
    ContentFilteringError,   # 内容被过滤
    APIConnectionError,      # 网络连接问题
    APITimeoutError,         # 请求超时
    InsufficientCreditsError,# 余额不足
    OverloadedError,         # 服务器过载
    ServerError,             # 通用服务器错误
)

try:
    result = await agent.call(str, "Do something")
except RateLimitError:
    await asyncio.sleep(60)
    result = await agent.call(str, "Do something")
except MaxTokensError:
    # 缩小Scope或提高限制
    pass
except ContentFilteringError:
    # 内容被过滤
    pass
except InferenceError as e:
    logger.error(f"Inference failed: {e}")
except AgenticaError as e:
    logger.error(f"SDK error: {e}")

Custom Exceptions

自定义异常

python
class DataValidationError(Exception):
    """Invalid input data."""
    pass

@agentic(DataValidationError)  # Pass exception type
async def analyze(data: str) -> dict:
    """
    Analyze data.

    Raises:
        DataValidationError: If data is malformed
    """
    ...

try:
    result = await analyze(raw_data)
except DataValidationError as e:
    logger.warning(f"Invalid: {e}")
python
class DataValidationError(Exception):
    """Invalid input data."""
    pass

@agentic(DataValidationError)  # 传入异常类型
async def analyze(data: str) -> dict:
    """
    Analyze data.

    Raises:
        DataValidationError: If data is malformed
    """
    ...

try:
    result = await analyze(raw_data)
except DataValidationError as e:
    logger.warning(f"Invalid: {e}")

Multi-Agent Patterns

多Agent模式

Custom Agent Class

自定义Agent类

python
from agentica.agent import Agent

class ResearchAgent:
    def __init__(self, web_search_fn):
        self._brain = Agent(
            premise="Research assistant.",
            scope={"web_search": web_search_fn}
        )

    async def research(self, topic: str) -> str:
        return await self._brain(str, f"Research: {topic}")

    async def summarize(self, text: str) -> str:
        return await self._brain(str, f"Summarize: {text}")
python
from agentica.agent import Agent

class ResearchAgent:
    def __init__(self, web_search_fn):
        self._brain = Agent(
            premise="Research assistant.",
            scope={"web_search": web_search_fn}
        )

    async def research(self, topic: str) -> str:
        return await self._brain(str, f"Research: {topic}")

    async def summarize(self, text: str) -> str:
        return await self._brain(str, f"Summarize: {text}")

Agent Orchestration

Agent编排

python
class LeadResearcher:
    def __init__(self):
        self._brain = Agent(
            premise="Coordinate research across subagents.",
            scope={"SubAgent": ResearchAgent}
        )

    async def __call__(self, query: str) -> str:
        return await self._brain(str, query)

lead = LeadResearcher()
report = await lead("Research AI agent frameworks 2025")
python
class LeadResearcher:
    def __init__(self):
        self._brain = Agent(
            premise="Coordinate research across subagents.",
            scope={"SubAgent": ResearchAgent}
        )

    async def __call__(self, query: str) -> str:
        return await self._brain(str, query)

lead = LeadResearcher()
report = await lead("Research AI agent frameworks 2025")

Tracing & Debugging

追踪与调试

OpenTelemetry Tracing

OpenTelemetry追踪

python
from agentica import initialize_tracing
python
from agentica import initialize_tracing

Initialize tracing (returns TracerProvider)

初始化追踪（返回TracerProvider）

tracer = initialize_tracing( service_name="my-agent-app", environment="development", # Optional tempo_endpoint="http://localhost:4317", # Optional: Grafana Tempo organization_id="my-org", # Optional log_level="INFO", # DEBUG, INFO, WARNING, ERROR instrument_httpx=False, # Optional: trace HTTP calls )
undefined
tracer = initialize_tracing( service_name="my-agent-app", environment="development", # 可选 tempo_endpoint="http://localhost:4317", # 可选：Grafana Tempo地址 organization_id="my-org", # 可选 log_level="INFO", # 可选值：DEBUG, INFO, WARNING, ERROR instrument_httpx=False, # 可选：追踪HTTP调用 )
undefined

SDK Debug Logging

SDK调试日志

python
from agentica import enable_sdk_logging
python
from agentica import enable_sdk_logging

Enable internal SDK logs (for debugging the SDK itself)

启用SDK内部日志（用于调试SDK本身）

disable_fn = enable_sdk_logging(log_tags="1")
disable_fn = enable_sdk_logging(log_tags="1")

... run agents ...

...运行Agent...

disable_fn() # Disable when done
undefined
disable_fn() # 完成后禁用
undefined

Top-Level Exports

顶层导出项

python
undefined
python
undefined

Main imports from agentica

从agentica导入的主要内容

from agentica import ( # Core Agent, # Synchronous agent class agentic, # @agentic decorator spawn, # Async agent creation
# Configuration
ModelStrings,       # Model string type hints
AgenticFunction,    # Agentic function type

# Token tracking
last_usage,         # Get last call's token usage
total_usage,        # Get cumulative token usage

# Tracing/Logging
initialize_tracing, # OpenTelemetry setup
enable_sdk_logging, # SDK debug logs

# Version
__version__,        # "0.3.1"
)
undefined
from agentica import ( # 核心类与函数 Agent, # 同步Agent类 agentic, # @agentic装饰器 spawn, # 异步Agent创建函数
# 配置相关
ModelStrings,       # 模型字符串类型提示
AgenticFunction,    # Agentic函数类型

# Token追踪
last_usage,         # 获取上次调用的Token使用量
total_usage,        # 获取累计Token使用量

# 追踪与日志
initialize_tracing, # OpenTelemetry初始化
enable_sdk_logging, # SDK调试日志启用

# 版本
__version__,        # 当前版本："0.3.1"
)
undefined

Checklist

检查清单

Before using Agentica:
  • Functions with 
    @agentic()
    MUST be 
    async
  • spawn()
    returns awaitable - use 
    await spawn(...)
  • agent.call()
    is awaitable - use 
    await agent.call(...)
  • First arg to 
    call()
    is return type, second is prompt string
  • Use 
    persist=True
    for conversation memory in 
    @agentic
  • Use 
    Agent()
    (not 
    spawn()
    ) in synchronous 
    __init__
  • Document exceptions in docstrings for agent to raise them
  • Import listeners from 
    agentica.logging.agent_listener
    (NOT 
    agentica.logging
    )
使用Agentica前请确认：
  • 被
    @agentic()
    装饰的函数必须是
    async
    函数
  • spawn()
    返回可等待对象 - 需使用
    await spawn(...)
  • agent.call()
    是可等待对象 - 需使用
    await agent.call(...)
  • call()
    的第一个参数是返回类型，第二个是提示词字符串
  • 为
    @agentic
    函数添加
    persist=True
    以实现对话记忆
  • 在同步
    __init__
    中使用
    Agent()
    而非
    spawn()
  • 在文档字符串中声明Agent需要抛出的异常
  • 从
    agentica.logging.agent_listener
    导入监听器（而非
    agentica.logging
    ）