agentica-sdk

Agentica SDK 参考（v0.3.1）

使用 Agentica 框架在 Python 中构建 AI Agent。Agent 可以实现函数、维护状态、使用工具，并相互协作。

适用场景

在以下情况下使用此技能：

• 构建新的 Python Agent
• 为现有代码添加 Agent 能力
• 将 MCP 工具与 Agent 集成
• 实现多 Agent 编排
• 调试 Agent 行为

快速开始

Agentic 函数（最简单）

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

生成 Agent（更多控制）

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

核心模式

返回类型

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

# Typed output
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
await agent.call(None, "Send message to John")

前提与系统提示词

# Premise: adds to default system prompt
agent = await spawn(premise="You are a math expert.")

# System: full control (replaces default)
agent = await spawn(system="You are a JSON-only responder.")

传递工具（作用域）

from agentica import agentic, spawn

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

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

# Per-call scope
result = await agent.call(
    dict[str, int],
    "Analyze the dataset",
    dataset=data,           # Available as 'dataset'
    analyzer=custom_fn      # Available as 'analyzer'
)

SDK 集成模式

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

Agent 实例化

spawn() - 异步（适用于大多数情况）

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

Agent() - 同步（用于 __init__）

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)

模型选择

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

# In decorator
@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 slug（例如，google/gemini-2.5-flash）

持久化（有状态 Agent）

@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

对于 spawn() Agent，同一实例在多次调用之间会自动保持状态。

Token 限制

from agentica import spawn, MaxTokens

# Simple limit
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
    )
)

Token 使用情况跟踪

from agentica import spawn, last_usage, total_usage

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

# Agent method
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()
async def my_fn(x: str) -> str: ...

await my_fn("test")
print(last_usage(my_fn))
print(total_usage(my_fn))

流式传输

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)
# chunk.role is 'user', 'agent', or 'system'

# Then await result
final = await result

MCP 集成

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 格式：

{
  "mcpServers": {
    "tavily-remote-mcp": {
      "command": "npx -y mcp-remote https://mcp.tavily.com/mcp/?tavilyApiKey=<key>",
      "env": {}
    }
  }
}

日志记录

默认行为

• 使用颜色输出到 stdout
• 写入 ./logs/agent-<id>.log

上下文日志记录

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

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

按 Agent 配置日志

# Listeners are in agent_listener submodule (NOT exported from 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
)

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

全局配置

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

错误处理

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}")

自定义异常

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}")

多 Agent 模式

自定义 Agent 类

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 编排

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

跟踪与调试

OpenTelemetry 跟踪

from agentica import initialize_tracing

# Initialize tracing (returns 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
)

SDK 调试日志

from agentica import enable_sdk_logging

# Enable internal SDK logs (for debugging the SDK itself)
disable_fn = enable_sdk_logging(log_tags="1")

# ... run agents ...

disable_fn()  # Disable when done

顶层导出

# Main imports from 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"
)

检查清单

使用 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 导入）