{"id":"agentica-sdk","name":"agentica-sdk","summary":"Agentica SDKでPythonエージェントを構築する - @agentic デコレーター、spawn()、永続性、MCP統合","body":"# Agentica SDK Reference (v0.3.1)\n\nBuild AI agents in Python using the Agentica framework. Agents can implement functions, maintain state, use tools, and coordinate with each other.\n\n## When to Use\n\nUse this skill when:\n- Building new Python agents\n- Adding agentic capabilities to existing code\n- Integrating MCP tools with agents\n- Implementing multi-agent orchestration\n- Debugging agent behavior\n\n## Quick Start\n\n### Agentic Function (simplest)\n\n```python\nfrom agentica import agentic\n\n@agentic()\nasync def add(a: int, b: int) -> int:\n    \"\"\"Returns the sum of a and b\"\"\"\n    ...\n\nresult = await add(1, 2)  # Agent computes: 3\n```\n\n### Spawned Agent (more control)\n\n```python\nfrom agentica import spawn\n\nagent = await spawn(premise=\"You are a truth-teller.\")\nresult: bool = await agent.call(bool, \"The Earth is flat\")\n# Returns: False\n```\n\n## Core Patterns\n\n### Return Types\n\n```python\n# String (default)\nresult = await agent.call(\"What is 2+2?\")\n\n# Typed output\nresult: int = await agent.call(int, \"What is 2+2?\")\nresult: dict[str, int] = await agent.call(dict[str, int], \"Count items\")\n\n# Side-effects only\nawait agent.call(None, \"Send message to John\")\n```\n\n### Premise vs System Prompt\n\n```python\n# Premise: adds to default system prompt\nagent = await spawn(premise=\"You are a math expert.\")\n\n# System: full control (replaces default)\nagent = await spawn(system=\"You are a JSON-only responder.\")\n```\n\n### Passing Tools (Scope)\n\n```python\nfrom agentica import agentic, spawn\n\n# In decorator\n@agentic(scope={'web_search': web_search_fn})\nasync def researcher(query: str) -> str:\n    \"\"\"Research a topic.\"\"\"\n    ...\n\n# In spawn\nagent = await spawn(\n    premise=\"Data analyzer\",\n    scope={\"analyze\": custom_analyzer}\n)\n\n# Per-call scope\nresult = await agent.call(\n    dict[str, int],\n    \"Analyze the dataset\",\n    dataset=data,           # Available as 'dataset'\n    analyzer=custom_fn      # Available as 'analyzer'\n)\n```\n\n### SDK Integration Pattern\n\n```python\nfrom slack_sdk import WebClient\n\nslack = WebClient(token=SLACK_TOKEN)\n\n# Extract specific methods\n@agentic(scope={\n    'list_users': slack.users_list,\n    'send_message': slack.chat_postMessage\n})\nasync def team_notifier(message: str) -> None:\n    \"\"\"Send team notifications.\"\"\"\n    ...\n```\n\n## Agent Instantiation\n\n### spawn() - Async (most cases)\n\n```python\nagent = await spawn(premise=\"Helpful assistant\")\n```\n\n### Agent() - Sync (for `__init__`)\n\n```python\nfrom agentica.agent import Agent\n\nclass CustomAgent:\n    def __init__(self):\n        # Synchronous - use Agent() not spawn()\n        self._brain = Agent(\n            premise=\"Specialized assistant\",\n            scope={\"tool\": some_tool}\n        )\n\n    async def run(self, task: str) -> str:\n        return await self._brain(str, task)\n```\n\n## Model Selection\n\n```python\n# In spawn\nagent = await spawn(\n    premise=\"Fast responses\",\n    model=\"openai:gpt-5\"  # Default: openai:gpt-4.1\n)\n\n# In decorator\n@agentic(model=\"anthropic:claude-sonnet-4.5\")\nasync def analyze(text: str) -> dict:\n    \"\"\"Analyze text.\"\"\"\n    ...\n```\n\n**Available models:**\n- `openai:gpt-3.5-turbo`, `openai:gpt-4o`, `openai:gpt-4.1`, `openai:gpt-5`\n- `anthropic:claude-sonnet-4`, `anthropic:claude-opus-4.1`\n- `anthropic:claude-sonnet-4.5`, `anthropic:claude-opus-4.5`\n- Any OpenRouter slug (e.g., `google/gemini-2.5-flash`)\n\n## Persistence (Stateful Agents)\n\n```python\n@agentic(persist=True)\nasync def chatbot(message: str) -> str:\n    \"\"\"Remembers conversation history.\"\"\"\n    ...\n\nawait chatbot(\"My name is Alice\")\nawait chatbot(\"What's my name?\")  # Knows: Alice\n```\n\nFor `spawn()` agents, state is automatic across calls to the same instance.\n\n## Token Limits\n\n```python\nfrom agentica import spawn, MaxTokens\n\n# Simple limit\nagent = await spawn(\n    premise=\"Brief responses\",\n    max_tokens=500\n)\n\n# Fine-grained control\nagent = await spawn(\n    premise=\"Controlled output\",\n    max_tokens=MaxTokens(\n        per_invocation=5000,  # Total across all rounds\n        per_round=1000,       # Per inference round\n        rounds=5              # Max inference rounds\n    )\n)\n```\n\n## Token Usage Tracking\n\n```python\nfrom agentica import spawn, last_usage, total_usage\n\nagent = await spawn(premise=\"You are helpful.\")\nawait agent.call(str, \"Hello!\")\n\n# Agent method\nusage = agent.last_usage()\nprint(f\"Last: {usage.input_tokens} in, {usage.output_tokens} out\")\n\nusage = agent.total_usage()\nprint(f\"Total: {usage.total_tokens} processed\")\n\n# For @agentic functions\n@agentic()\nasync def my_fn(x: str) -> str: ...\n\nawait my_fn(\"test\")\nprint(last_usage(my_fn))\nprint(total_usage(my_fn))\n```\n\n## Streaming\n\n```python\nfrom agentica import spawn\nfrom agentica.logging.loggers import StreamLogger\nimport asyncio\n\nagent = await spawn(premise=\"You are helpful.\")\n\nstream = StreamLogger()\nwith stream:\n    result = asyncio.create_task(\n        agent.call(bool, \"Is Paris the capital of France?\")\n    )\n\n# Consume stream FIRST for live output\nasync for chunk in stream:\n    print(chunk.content, end=\"\", flush=True)\n# chunk.role is 'user', 'agent', or 'system'\n\n# Then await result\nfinal = await result\n```\n\n## MCP Integration\n\n```python\nfrom agentica import spawn, agentic\n\n# Via config file\nagent = await spawn(\n    premise=\"Tool-using agent\",\n    mcp=\"path/to/mcp_config.json\"\n)\n\n@agentic(mcp=\"path/to/mcp_config.json\")\nasync def tool_user(query: str) -> str:\n    \"\"\"Uses MCP tools.\"\"\"\n    ...\n```\n\n**mcp_config.json format:**\n```json\n{\n  \"mcpServers\": {\n    \"tavily-remote-mcp\": {\n      \"command\": \"npx -y mcp-remote https://mcp.tavily.com/mcp/?tavilyApiKey=<key>\",\n      \"env\": {}\n    }\n  }\n}\n```\n\n## Logging\n\n### Default Behavior\n- Prints to stdout with colors\n- Writes to `./logs/agent-<id>.log`\n\n### Contextual Logging\n\n```python\nfrom agentica.logging.loggers import FileLogger, PrintLogger\nfrom agentica.logging.agent_logger import NoLogging\n\n# File only\nwith FileLogger():\n    agent = await spawn(premise=\"Debug agent\")\n    await agent.call(int, \"Calculate\")\n\n# Silent\nwith NoLogging():\n    agent = await spawn(premise=\"Silent agent\")\n```\n\n### Per-Agent Logging\n\n```python\n# Listeners are in agent_listener submodule (NOT exported from agentica.logging)\nfrom agentica.logging.agent_listener import (\n    PrintOnlyListener,  # Console output only\n    FileOnlyListener,   # File logging only\n    StandardListener,   # Both console + file (default)\n    NoopListener,       # Silent - no logging\n)\n\nagent = await spawn(\n    premise=\"Custom logging\",\n    listener=PrintOnlyListener\n)\n\n# Silent agent\nagent = await spawn(\n    premise=\"Silent agent\",\n    listener=NoopListener\n)\n```\n\n### Global Config\n\n```python\nfrom agentica.logging.agent_listener import (\n    set_default_agent_listener,\n    get_default_agent_listener,\n    PrintOnlyListener,\n)\n\nset_default_agent_listener(PrintOnlyListener)\nset_default_agent_listener(None)  # Disable all\n```\n\n## Error Handling\n\n```python\nfrom agentica.errors import (\n    AgenticaError,           # Base for all SDK errors\n    RateLimitError,          # Rate limiting\n    InferenceError,          # HTTP errors from inference\n    MaxTokensError,          # Token limit exceeded\n    MaxRoundsError,          # Max inference rounds exceeded\n    ContentFilteringError,   # Content filtered\n    APIConnectionError,      # Network issues\n    APITimeoutError,         # Request timeout\n    InsufficientCreditsError,# Out of credits\n    OverloadedError,         # Server overloaded\n    ServerError,             # Generic server error\n)\n\ntry:\n    result = await agent.call(str, \"Do something\")\nexcept RateLimitError:\n    await asyncio.sleep(60)\n    result = await agent.call(str, \"Do something\")\nexcept MaxTokensError:\n    # Reduce scope or increase limits\n    pass\nexcept ContentFilteringError:\n    # Content was filtered\n    pass\nexcept InferenceError as e:\n    logger.error(f\"Inference failed: {e}\")\nexcept AgenticaError as e:\n    logger.error(f\"SDK error: {e}\")\n```\n\n### Custom Exceptions\n\n```python\nclass DataValidationError(Exception):\n    \"\"\"Invalid input data.\"\"\"\n    pass\n\n@agentic(DataValidationError)  # Pass exception type\nasync def analyze(data: str) -> dict:\n    \"\"\"\n    Analyze data.\n\n    Raises:\n        DataValidationError: If data is malformed\n    \"\"\"\n    ...\n\ntry:\n    result = await analyze(raw_data)\nexcept DataValidationError as e:\n    logger.warning(f\"Invalid: {e}\")\n```\n\n## Multi-Agent Patterns\n\n### Custom Agent Class\n\n```python\nfrom agentica.agent import Agent\n\nclass ResearchAgent:\n    def __init__(self, web_search_fn):\n        self._brain = Agent(\n            premise=\"Research assistant.\",\n            scope={\"web_search\": web_search_fn}\n        )\n\n    async def research(self, topic: str) -> str:\n        return await self._brain(str, f\"Research: {topic}\")\n\n    async def summarize(self, text: str) -> str:\n        return await self._brain(str, f\"Summarize: {text}\")\n```\n\n### Agent Orchestration\n\n```python\nclass LeadResearcher:\n    def __init__(self):\n        self._brain = Agent(\n            premise=\"Coordinate research across subagents.\",\n            scope={\"SubAgent\": ResearchAgent}\n        )\n\n    async def __call__(self, query: str) -> str:\n        return await self._brain(str, query)\n\nlead = LeadResearcher()\nreport = await lead(\"Research AI agent frameworks 2025\")\n```\n\n## Tracing & Debugging\n\n### OpenTelemetry Tracing\n\n```python\nfrom agentica import initialize_tracing\n\n# Initialize tracing (returns TracerProvider)\ntracer = initialize_tracing(\n    service_name=\"my-agent-app\",\n    environment=\"development\",  # Optional\n    tempo_endpoint=\"http://localhost:4317\",  # Optional: Grafana Tempo\n    organization_id=\"my-org\",  # Optional\n    log_level=\"INFO\",  # DEBUG, INFO, WARNING, ERROR\n    instrument_httpx=False,  # Optional: trace HTTP calls\n)\n```\n\n### SDK Debug Logging\n\n```python\nfrom agentica import enable_sdk_logging\n\n# Enable internal SDK logs (for debugging the SDK itself)\ndisable_fn = enable_sdk_logging(log_tags=\"1\")\n\n# ... run agents ...\n\ndisable_fn()  # Disable when done\n```\n\n## Top-Level Exports\n\n```python\n# Main imports from agentica\nfrom agentica import (\n    # Core\n    Agent,              # Synchronous agent class\n    agentic,            # @agentic decorator\n    spawn,              # Async agent creation\n\n    # Configuration\n    ModelStrings,       # Model string type hints\n    AgenticFunction,    # Agentic function type\n\n    # Token tracking\n    last_usage,         # Get last call's token usage\n    total_usage,        # Get cumulative token usage\n\n    # Tracing/Logging\n    initialize_tracing, # OpenTelemetry setup\n    enable_sdk_logging, # SDK debug logs\n\n    # Version\n    __version__,        # \"0.3.1\"\n)\n```\n\n## Checklist\n\nBefore using Agentica:\n- [ ] Functions with `@agentic()` MUST be `async`\n- [ ] `spawn()` returns awaitable - use `await spawn(...)`\n- [ ] `agent.call()` is awaitable - use `await agent.call(...)`\n- [ ] First arg to `call()` is return type, second is prompt string\n- [ ] Use `persist=True` for conversation memory in `@agentic`\n- [ ] Use `Agent()` (not `spawn()`) in synchronous `__init__`\n- [ ] Document exceptions in docstrings for agent to raise them\n- [ ] Import listeners from `agentica.logging.agent_listener` (NOT `agentica.logging`)","author":"@parcadei","ownerProfile":null,"authorContacts":null,"sourceUrl":"https://github.com/parcadei/Continuous-Claude-v3/tree/main/.claude/skills/agentica-sdk","license":"MIT","category":"coding","lang":"en","tokens":2799,"stars":0,"calls30d":1,"claimed":false,"visibility":"public","origin":"crawler","version":"0.1.0","createdAt":"2026-08-22","updatedAt":"2026-08-22","files":[],"requires":{"mcp":[],"tools":["Bash","Read","Write","Edit"]},"safety":{"flags":[],"scannedAt":"2026-08-22","hasScripts":false,"networkEndpoints":["mcp.tavily.com"]}}