Skip to content

Architecture 2 Agents

Jump to bottom
Danny Volz edited this page Jul 9, 2025 · 3 revisions

Agents Module Architecture

Code References

Overview

The Agents module provides an autonomous AI agent framework with tool-calling capabilities. It allows AI models to execute functions and interact with external systems through a unified interface. The main entry point is the Agent class in coda/agents/agent.py.

Module Structure

coda/agents/
├── __init__.py          # Module exports (verified)
├── agent.py             # Core Agent implementation (verified)
├── decorators.py        # @tool decorator (verified)
├── function_tool.py     # FunctionTool wrapper class (verified)
├── tool_adapter.py      # MCP tool integration (verified)
├── builtin_tools.py     # Pre-built tool functions (verified)
└── types.py            # Type definitions (verified)

Key Components

Component 1: Agent Core

Location: coda/agents/agent.py

Purpose: Main agent class that orchestrates AI conversations with tool execution

Key Methods:

  • __init__(): Initializes agent with provider, model, and tools
  • run_async(): Executes agent with tool calling support
  • run_async_streaming(): Streaming execution with real-time output
  • _handle_tool_calls(): Orchestrates tool execution

Component 2: Tool Decorator

Location: coda/agents/decorators.py

Purpose: Decorator that marks functions as agent-callable tools

Implementation:

def tool(func=None, *, name=None, description=None):
    """Decorator to mark a function as a tool."""
    def decorator(f):
        f._is_tool = True
        f._tool_name = name or f.__name__
        f._tool_description = description or f.__doc__ or ""

Component 3: FunctionTool

Location: coda/agents/function_tool.py

Purpose: Wrapper that converts Python functions into tool specifications

Key Methods:

  • from_callable(): Factory method to create from decorated functions
  • execute(): Executes the wrapped function with arguments
  • _build_parameters(): Generates JSON Schema from function signature

Component 4: MCP Tool Adapter

Location: coda/agents/tool_adapter.py

Purpose: Adapts MCP (Model Context Protocol) tools for use with agents

Key Methods:

  • convert_mcp_tool(): Converts MCP tool to FunctionTool
  • get_all_tools(): Retrieves all registered MCP tools

Implementation Details

Design Patterns Used

Decorator Pattern

Implementation: coda/agents/decorators.py

The @tool decorator adds metadata to functions:

@tool(description="Read contents of a file")
def read_file(path: str) -> str:
    """Read and return the contents of a file."""
    with open(path, 'r') as f:
        return f.read()

Adapter Pattern

Implementation: coda/agents/tool_adapter.py

Adapts MCP tools to the agent's FunctionTool interface:

def convert_mcp_tool(self, mcp_tool: Type[BaseTool]) -> FunctionTool:
    """Convert an MCP tool to a FunctionTool."""
    # Creates wrapper function that calls MCP tool
    async def tool_wrapper(**kwargs):
        return await tool_instance.execute(**kwargs)

Factory Pattern

Implementation: coda/agents/function_tool.py

@classmethod
def from_callable(cls, func: Callable) -> "FunctionTool":
    """Create a FunctionTool from a callable."""
    if not hasattr(func, "_is_tool"):
        raise ValueError(f"Function {func.__name__} is not decorated with @tool")

Data Flow

sequenceDiagram
    participant User
    participant Agent
    participant Provider
    participant FunctionTool
    participant Tool Function
    
    User->>Agent: run_async(messages)
    Agent->>Provider: Send messages with tools
    Provider->>Provider: Generate response
    
    alt Tool calls requested
        Provider->>Agent: Response with tool_calls
        loop For each tool call
            Agent->>Agent: Create RequiredAction
            Agent->>FunctionTool: execute(arguments)
            FunctionTool->>Tool Function: Call with args
            Tool Function->>FunctionTool: Return result
            FunctionTool->>Agent: Return PerformedAction
        end
        Agent->>Agent: Add tool results to messages
        Agent->>Provider: Continue with results
    end
    
    Provider->>Agent: Final response
    Agent->>User: Return RunResponse
Loading

Class Hierarchy

classDiagram
    BaseModel <|-- FunctionTool
    BaseModel <|-- RequiredAction
    BaseModel <|-- PerformedAction
    BaseModel <|-- RunResponse
    
    class Agent {
        -provider: BaseProvider
        -model: str
        -tools: List[FunctionTool]
        +run_async(messages)
        +run_async_streaming(messages)
        -_handle_tool_calls(tool_calls)
    }
    
    class FunctionTool {
        +name: str
        +description: str
        +parameters: dict
        +callable: Callable
        +from_callable(func)
        +execute(arguments)
    }
    
    class MCPToolAdapter {
        +convert_mcp_tool(mcp_tool)
        +get_all_tools()
    }
    
    note for Agent "Defined in agent.py"
    note for FunctionTool "Defined in function_tool.py"
Loading

API Reference

Public Functions

@tool decorator

Location: coda/agents/decorators.py

Purpose: Marks a function as an agent-callable tool

Usage:

@tool  # Simple usage
def my_tool(): ...

@tool(name="custom", description="Custom tool")  # With metadata
def another_tool(): ...

Public Classes

Agent

Location: coda/agents/agent.py

Purpose: Autonomous AI agent with tool-calling capabilities

Constructor:

def __init__(
    self,
    provider: BaseProvider,
    model: str,
    instructions: str = "",
    tools: List[Union[Callable, FunctionTool]] = None,
    name: str = None
)

Key Methods:

  • run_async(messages, instructions): Async execution
  • run_async_streaming(messages, instructions): Streaming execution
  • run(messages, instructions): Sync wrapper
  • as_tool(): Convert agent to tool

FunctionTool

Location: coda/agents/function_tool.py

Purpose: Wrapper for Python functions with tool metadata

Fields:

  • name (str): Tool name
  • description (str): Tool description
  • parameters (dict): JSON Schema parameters
  • callable (Callable): Wrapped function

Configuration

Tools are configured through:

  1. Decorator parameters: @tool(name="...", description="...")
  2. Function docstrings: Used as default descriptions
  3. Type hints: Used to generate parameter schemas

Dependencies

Internal Dependencies

  • coda.providers.base: Provider interface
  • coda.tools.base: Tool registry for MCP tools
  • coda.tools.executor: Tool execution permissions

External Dependencies

  • pydantic>=2.0: Data validation and models
  • rich>=13.7.0: Console output formatting
  • typing_extensions: Enhanced typing support

Testing

Test Coverage

  • Unit tests: tests/test_agent.py
  • Tool calling tests: tests/test_tool_calling.py
  • Integration tests: Various provider-specific tests

Key Test Cases

  1. Basic Agent Execution: Tests agent without tools
  2. Tool Calling: Tests tool execution flow
  3. Streaming Support: Tests streaming responses

Error Handling

Exception Handling

Tool execution errors are captured and returned to the model:

try:
    result = tool.execute(filtered_args)
except Exception as e:
    logger.error(f"Tool execution error: {e}")
    result = f"Error: {str(e)}"

Loop Detection

Implementation: agent.py

Prevents infinite tool-calling loops by tracking call patterns and breaking after repeated sequences.

Performance Considerations

  1. Async Execution: All operations are async-first
  2. Streaming Support: Reduces time-to-first-token
  3. Tool Map Caching: O(1) tool lookup
  4. Lazy Tool Processing: Tools converted only when needed

Security Considerations

  1. Tool Permissions: Integrated with permission system
  2. Argument Filtering: Only declared parameters passed to tools
  3. Error Isolation: Tool errors don't crash agent

Examples

Basic Usage

from coda.agents import Agent, tool

@tool
def get_weather(location: str) -> str:
    """Get weather for a location."""
    return f"Sunny in {location}"

agent = Agent(
    provider=provider,
    model="gpt-4",
    tools=[get_weather]
)

response = await agent.run_async([
    {"role": "user", "content": "What's the weather in Paris?"}
])

Advanced Usage

@tool(description="Fetch data from a URL asynchronously")
async def fetch_data(url: str) -> str:
    """Fetch and return data from a URL."""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

Integration Points

With Other Modules

  1. Provider Module: Uses provider for AI completions
  2. Tools Module: Integrates MCP tools via adapter
  3. CLI Module: Used in agent_chat.py for interactive sessions

Extension Points

  1. Custom Tools: Create with @tool decorator
  2. Tool Adapters: Implement adapters for other tool protocols
  3. Agent Subclassing: Extend Agent for specialized behavior

Known Limitations

  1. Tool Discovery: Tools must be explicitly passed to agent
  2. Parameter Types: Complex types may not serialize correctly

References

Related Documentation

Source Files

All files referenced in this document:

Clone this wiki locally