Skip to content

Architecture

Timestep is built around the A2A (Agent-to-Agent) and MCP (Model Context Protocol) protocols, providing a clean foundation for building multi-agent systems. This document explains the architecture and how these protocols work together.

Overview

Timestep follows the Task-generating Agents philosophy from the A2A Protocol, where agents always respond with Task objects. We use MCP sampling to enable seamless agent-to-agent handoffs.

A2A Protocol Integration

Task-Generating Agents

Following the A2A Task-generating Agents philosophy, our agents always respond with Task objects (never just Messages). This provides:

  • State management: Tasks can be in various states (created, input-required, completed, canceled, etc.)
  • Progress tracking: Task status updates provide visibility into agent progress
  • Multi-turn interactions: Tasks support context IDs for grouping related interactions
  • Structured tool communication: Tool calls are communicated via input-required state with DataPart

A2A Server Architecture

The A2A server is implemented by the Agent class, which contains a Loop (AgentExecutor):

# lib/python/core/agent.py
from timestep.core import Agent

# Agent always responds with Task objects
agent = Agent(
    agent_id="agent-1",
    name="My Agent",
    model="gpt-4o-mini",
    context_id_to_environment_uri={
        "context-1": "http://localhost:8080/mcp"
    }
)

The Loop class (inside Agent) implements the AgentExecutor interface:

# lib/python/core/loop/__main__.py
class Loop(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # Process message and generate response
        # Always emits Task objects via event_queue
        # Manages task lifecycle and state transitions
        # Includes tool calls in DataPart when input-required

Key characteristics: - Agent class wraps A2A SDK's AgentExecutor (Loop) - Always responds with Task objects - Manages task lifecycle and state transitions - Includes tool calls in DataPart when input-required

A2A input-required with DataPart

When an agent needs to call tools, it uses A2A's input-required state with a DataPart:

  1. Agent sets task state to input-required
  2. Agent includes DataPart in task status message parts: 
    DataPart(
    data={
        "tool_calls": [
            {
                "function": {
                    "name": "handoff",
                    "arguments": json.dumps({
                        "agent_uri": "http://.../agents/...",
                        "message": "What's the weather in Oakland?"
                    })
                }
            }
        ]
    }
)
  3. Client detects input-required state
  4. Client extracts tool calls from DataPart.data.tool_calls
  5. Client executes tools via MCP
  6. Client sends results back to A2A server

This pattern allows agents to communicate tool needs in a structured, protocol-compliant way.

MCP Protocol Integration

MCP Server Architecture

The MCP server is implemented by the Environment class, which extends FastMCP:

# lib/python/core/environment.py
from timestep.core import Environment

# Create environment (MCP Server)
environment = Environment(
    environment_id="env-1",
    context_id="context-1",
    agent_id="agent-1",
    enable_handoff=True,  # Enable built-in handoff tool (default)
)

# Register custom tools
@environment.tool()
async def my_tool(param: str) -> dict:
    """Tool description."""
    return {"result": param}

The built-in handoff tool is automatically registered when enable_handoff=True:

# Built into Environment class
@environment.tool()  # Conditionally registered based on enable_handoff
async def handoff(
    agent_uri: str,
    message: str,
    ctx: Context = None,
) -> Dict[str, Any]:
    """Handoff tool using MCP sampling."""
    result = await ctx.session.create_message(
        messages=[SamplingMessage(...)],
        metadata={"agent_uri": agent_uri}
    )
    return {"response": result.content.text.strip()}

Key characteristics: - Environment extends FastMCP and provides MCP server functionality - Built-in handoff tool is registered by default (controlled by enable_handoff parameter) - Custom tools registered with @environment.tool() decorator - handoff tool uses sampling to trigger A2A requests - HTTP transport for client connections - Supports both tool execution and sampling

MCP Sampling for Handoffs

MCP's sampling feature enables server-initiated LLM interactions. We use it for handoffs:

Flow: 1. Agent calls MCP handoff tool with target agent URI 2. MCP server calls client's sampling callback via ctx.session.create_message() 3. Sampling callback (in client) makes A2A request to target agent 4. Target agent processes and responds 5. Response returned to MCP server 6. MCP server returns result to original agent

Implementation: 

# In client
async def mcp_sampling_callback(
    context: RequestContext,
    params: CreateMessageRequestParams,
) -> CreateMessageResult:
    # Extract agent_uri from metadata
    agent_uri = params.metadata.get("agent_uri")
    message_text = params.messages[0].content.text

    # Make A2A request to target agent
    result_text = await handle_agent_handoff(agent_uri, message_text)

    return CreateMessageResult(
        role="assistant",
        content=TextContent(type="text", text=result_text)
    )

This pattern allows agents to delegate work to specialized peers without requiring direct A2A client capabilities in the MCP server.

Client Architecture

The Loop component provides a /v1/responses endpoint that orchestrates A2A and MCP interactions:

# lib/python/core/loop/__main__.py
from timestep.core import Loop

# Create Loop instance
loop = Loop(
    agent=agent,
    agent_base_url="http://localhost:8000",
    context_id_to_environment_uri={
        "context-1": "http://localhost:8080/mcp"
    }
)

# Mount Loop routes to FastAPI app
for route in loop.fastapi_app.routes:
    fastapi_app.add_api_route(route.path, route.endpoint, methods=route.methods)

The Loop component handles: - Receiving requests via /v1/responses endpoint - Converting to A2A messages and sending to agent - Monitoring A2A task state transitions - Extracting tool calls from DataPart when input-required - Executing tools via MCP (including built-in handoff execution) - Sending tool results back to A2A server - Converting A2A responses back to /v1/responses format

For custom clients, see scripts/personal_assistant_test_client.py for examples of direct A2A client usage.

Complete Interaction Flow

Here's a complete example of a handoff interaction:

  1. User sends message to Personal Assistant Agent via A2A client
  2. A2A server creates Task and processes message
  3. Agent determines it needs weather information
  4. Agent sets task state to input-required with DataPart containing handoff tool call
  5. Client detects input-required state
  6. Client extracts handoff tool call from DataPart
  7. Client calls MCP handoff tool with weather agent URI
  8. MCP server calls client's sampling callback
  9. Sampling callback makes A2A request to weather agent
  10. Weather agent processes and responds with weather data
  11. Response returned to MCP server
  12. MCP server returns result to client
  13. Client sends result back to A2A server as user message
  14. Personal assistant receives weather data and presents to user
  15. Task state transitions to completed

Protocol Specifications

Key Design Decisions

Why Task-Generating Agents?

  • Provides clear state management
  • Enables progress tracking
  • Supports multi-turn interactions
  • Structured tool call communication

Why MCP Sampling for Handoffs?

  • Allows agents to delegate without direct A2A client capabilities
  • Leverages MCP's built-in sampling mechanism
  • Keeps handoff logic in the client (where A2A client exists)
  • Maintains separation of concerns

Why DataPart for Tool Calls?

  • Protocol-compliant way to communicate tool needs
  • Structured format for tool call information
  • Supports multiple tool calls in single message
  • Clear separation between content and tool calls

Cross-Language Parity

Python and TypeScript implementations: - Use same A2A and MCP protocol patterns - Follow same Task-generating Agent philosophy - Use same input-required with DataPart pattern - Implement same handoff flow via MCP sampling - Compatible A2A Task and MCP message formats