Pattern 1: Pydantic AI Agent Definition
Agents are module-level singletons. Tools and validators use @agent.tool / @agent.output_validator.
# agents/my_agent.py
import logging
from pydantic_ai import Agent, RunContext
from second_brain.deps import BrainDeps
from second_brain.schemas import MyResult
logger = logging.getLogger(__name__)
my_agent = Agent(
deps_type=BrainDeps,
output_type=MyResult,
retries=3,
instructions="You are a ... agent. Always ...",
)
@my_agent.tool
async def my_tool(ctx: RunContext[BrainDeps], query: str) -> str:
"""Tool description shown to the LLM."""
try:
result = await ctx.deps.storage_service.get_something(query)
return result or "No results found."
except Exception as e:
return tool_error("my_tool", e)Pattern 2: MCP Tool (server.py)
Every MCP tool: validate input → get deps → run with timeout → format as markdown string.
@server.tool()
async def my_mcp_tool(query: str) -> str:
"""Tool description for Claude. Args: query: What to search for."""
try:
query = _validate_mcp_input(query, label="query")
except ValueError as e:
return str(e)
deps = _get_deps()
model = _get_model()
timeout = deps.config.api_timeout_seconds
try:
async with asyncio.timeout(timeout):
result = await my_agent.run(query, deps=deps, model=model)
except TimeoutError:
return f"Timed out after {timeout}s."
return f"# Result\n{result.output.answer}"Pattern 3: Service Class
Services accept BrainConfig, initialize clients in __init__, expose async methods.
# services/my_service.py
import logging
from second_brain.config import BrainConfig
logger = logging.getLogger(__name__)
class MyService:
"""One-line description."""
def __init__(self, config: BrainConfig):
self.config = config
self.client = SomeClient(api_key=config.some_api_key)
async def get_data(self, query: str) -> list[dict]:
try:
return await self.client.search(query, limit=self.config.memory_search_limit)
except Exception as e:
logger.warning("MyService.get_data failed: %s", e)
return []