agentica-sdk

Agentica SDK リファレンス（v0.3.1）

Agentica フレームワークを使用して、Python で AI エージェントを構築します。エージェントは関数を実装し、状態を保持し、ツールを使用して、相互に連携できます。

使用するタイミング

次の場合に、このスキルを使用します。

• 新しい Python エージェントを構築する
• 既存のコードにエージェント機能を追加する
• MCP ツールをエージェントと連携させる
• マルチエージェントのオーケストレーションを実装する
• エージェントの動作をデバッグする

クイックスタート

エージェント関数（最もシンプル）

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

起動したエージェント（より細かく制御）

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

エージェントのインスタンス化

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 スラッグ（例：google/gemini-2.5-flash）

永続化（状態を持つエージェント）

@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() で作成したエージェントの場合、同じインスタンスへの呼び出し間で状態が自動的に保持されます。

トークン上限

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

トークン使用量の追跡

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

エージェントごとのロギング

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

マルチエージェントのパターン

カスタムエージェントクラス

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

エージェントのオーケストレーション

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 可能な値を返します。await spawn(...) を使用してください
• agent.call() は await 可能です。await agent.call(...) を使用してください
• call() の第 1 引数は戻り値の型、第 2 引数はプロンプト文字列です
• @agentic で会話のメモリーを保持するには、persist=True を使用してください
• 同期処理の __init__ では spawn() ではなく Agent() を使用してください
• エージェントに例外を発生させるには、docstring に例外を記載してください
• リスナーは agentica.logging ではなく、agentica.logging.agent_listener からインポートしてください