A2a Mastra

A2A Mastra Demo - Multi-Agent System with Amazon Bedrock

A demonstration of Agent-to-Agent (A2A) communication protocol using the Mastra framework, featuring multiple specialized AI agents powered by Amazon Bedrock. This project showcases how autonomous agents can communicate, collaborate, and delegate tasks to achieve complex goals.

The system uses a hybrid architecture where Gateway agent(A2A Client Agent) runs on Express server with REST API implementation, while Data Processor, Summarizer, and Web Search agents use Mastra Dev Server with native A2A protocol.

🏗️ Architecture Overview

The system consists of four specialized agents that communicate via the A2A protocol

1. Gateway Agent - Request routing and workflow orchestration(A2A Client Agent)
2. Data Processor Agent - Data analysis and transformation(A2A Remote Agent)
3. Summarizer Agent - Content summarization and insight extraction(A2A Remote Agent)
4. Web Search Agent - Real-time web information retrieval(A2A Remote Agent)

Technology Stack

• Framework: Hybrid Architecture - Gateway (Express Server + REST API) + Data Processor, Summarizer, Web Search (Mastra Dev Server)
• LLM: Amazon Bedrock Claude 3.5 Sonnet
• Language: TypeScript
• Frontend: Next.js
• Containerization: Docker & Docker Compose
• Observability: Langfuse
• Web Search: Brave Search API + MCP (Model Context Protocol)

System Architecture Overview

flowchart TB
   subgraph subGraph0["Express + REST API - Gateway"]
      GW_EXPRESS["Express ServerPort: 3001"]
      GW_ROUTES["Express Routes/api/gateway/*"]
      GW_MASTRA_AGENT["Gateway Agent"]
      GW_REST_API["REST API Interface"]
   end
   subgraph subGraph1["Gateway Agent (Express Server)"]
      subGraph0
   end
   subgraph subGraph2["Express Server Agent"]
      subGraph1
   end
   subgraph subGraph3["Mastra Native Stack - Data Processor"]
      DP_MASTRA_DEV["Mastra Dev ServerPort: 3002"]
      DP_MASTRA_HONO["Built-in Hono Server"]
      DP_MASTRA_AGENT["Data Processor Agent"]
      DP_MASTRA_A2A["Native A2A Protocol"]
      DP_MASTRA_STORAGE["In-Memory StorageLibSQL"]
   end
   subgraph subGraph4["Data Processor Agent (Mastra Dev Server)"]
      subGraph3
   end
   subgraph subGraph5["Mastra Native Stack - Summarizer"]
      SM_MASTRA_DEV["Mastra Dev ServerPort: 3003"]
      SM_MASTRA_HONO["Built-in Hono Server"]
      SM_MASTRA_AGENT["Summarizer Agent"]
      SM_MASTRA_A2A["Native A2A Protocol"]
      SM_MASTRA_STORAGE["In-Memory StorageLibSQL"]
   end
   subgraph subGraph6["Summarizer Agent (Mastra Dev Server)"]
      subGraph5
   end
   subgraph subGraph7["Mastra Native Stack - Web Search"]
      WS_MASTRA_DEV["Mastra Dev ServerPort: 3004"]
      WS_MASTRA_HONO["Built-in Hono Server"]
      WS_MASTRA_AGENT["Web Search Agent"]
      WS_MASTRA_A2A["Native A2A Protocol"]
      WS_MASTRA_STORAGE["In-Memory StorageLibSQL"]
      WS_MCP_INTEGRATION["MCP Integration"]
   end
   subgraph subGraph8["Web Search Agent (Mastra Dev Server)"]
      subGraph7
   end
   subgraph subGraph9["Mastra Dev Server Agents"]
      subGraph4
      subGraph6
      subGraph8
   end
   subgraph subGraph10["External Services"]
      BEDROCK["Amazon BedrockClaude 3.5 Sonnet"]
      LANGFUSE["LangfuseObservability"]
      BRAVE["Brave Search API"]
      MCP["MCP ServerWeb Search Tools"]
   end
   GW_EXPRESS --> GW_ROUTES
   GW_ROUTES --> GW_MASTRA_AGENT
   GW_MASTRA_AGENT --> GW_REST_API & BEDROCK
   DP_MASTRA_DEV --> DP_MASTRA_HONO & DP_MASTRA_STORAGE
   DP_MASTRA_HONO --> DP_MASTRA_AGENT
   DP_MASTRA_AGENT --> DP_MASTRA_A2A & BEDROCK
   SM_MASTRA_DEV --> SM_MASTRA_HONO & SM_MASTRA_STORAGE
   SM_MASTRA_HONO --> SM_MASTRA_AGENT
   SM_MASTRA_AGENT --> SM_MASTRA_A2A & BEDROCK
   WS_MASTRA_DEV --> WS_MASTRA_HONO & WS_MASTRA_STORAGE
   WS_MASTRA_HONO --> WS_MASTRA_AGENT
   WS_MASTRA_AGENT --> WS_MASTRA_A2A & WS_MCP_INTEGRATION & BEDROCK
   GW_REST_API  DP_MASTRA_A2A & SM_MASTRA_A2A & WS_MASTRA_A2A
   WS_MCP_INTEGRATION -- MCP Stdio --> MCP
   MCP --> BRAVE
   GW_MASTRA_AGENT -. Traces .-> LANGFUSE
   DP_MASTRA_AGENT -. Traces .-> LANGFUSE
   SM_MASTRA_AGENT -. Traces .-> LANGFUSE
   WS_MASTRA_AGENT -. Traces .-> LANGFUSE

   style GW_EXPRESS fill:#e3f2fd
   style DP_MASTRA_DEV fill:#e8f5e8
   style DP_MASTRA_STORAGE fill:#fff3e0
   style SM_MASTRA_DEV fill:#e8f5e8
   style SM_MASTRA_STORAGE fill:#fff3e0
   style WS_MASTRA_DEV fill:#e8f5e8
   style WS_MASTRA_STORAGE fill:#fff3e0

Architecture Features:

• Hybrid Implementation: Gateway uses Express with REST API, Data Processor/Summarizer/Web Search use Mastra Dev Server
• Mixed Communication Protocols: REST API for Gateway, native Mastra A2A for other agents
• Native Integration: Data Processor, Summarizer, and Web Search agents use built-in Hono server and LibSQL storage
• MCP Support: Web Search agent integrates MCP protocol for external tool access

Simplified System Architecture

graph TB
   subgraph "Frontend Layer"
      UI[Next.js FrontendPort: 3000]
   end

   subgraph "Agent Layer - Hybrid Architecture"
      GW[Gateway AgentExpress ServerPort: 3001]
      DP[Data ProcessorMastra Dev ServerPort: 3002]
      SM[Summarizer AgentMastra Dev ServerPort: 3003]
      WS[Web Search AgentMastra Dev ServerPort: 3004]
   end

   subgraph "External Services"
      BEDROCK[Amazon BedrockClaude 3.5 Sonnet]
      LANGFUSE[LangfuseTracing]
      BRAVE[Brave Search API]
      MCP[MCP ServerWeb Search Tools]
   end

   UI -->|HTTP/REST| GW
   GW |Native A2A Protocol| DP
   GW |Native A2A Protocol| SM
   GW |Native A2A Protocol| WS

   DP --> BEDROCK
   SM --> BEDROCK
   GW --> BEDROCK
   WS --> BEDROCK
   WS --> |MCP Stdio| MCP
   MCP --> BRAVE

   GW -.->|Traces| LANGFUSE
   DP -.->|Traces| LANGFUSE
   SM -.->|Traces| LANGFUSE
   WS -.->|Traces| LANGFUSE

   style UI fill:#e1f5fe
   style GW fill:#e3f2fd
   style DP fill:#e8f5e8
   style SM fill:#e8f5e8
   style WS fill:#e8f5e8

🚀 Features

• Hybrid API Communication: REST API for Gateway, native Mastra A2A for Data Processor/Summarizer/Web Search
• Mixed Architecture: Express server for Gateway, Mastra Dev Server for Data Processor/Summarizer/Web Search
• Workflow Orchestration: Complex multi-step workflows with automatic task delegation
• Real-time Visualization: Live visualization of agent communication flows
• Tracing & Observability: Comprehensive tracing with Langfuse integration
• MCP Integration: Model Context Protocol support for web search capabilities
• Japanese Language Support: All agents respond in Japanese
• Containerized Deployment: Docker-based microservices architecture

📋 Prerequisites

• Docker and Docker Compose
• Node.js 22+ (for local development)
• AWS Account with Bedrock access
• Langfuse account (optional, for tracing)
• Brave Search API key (optional, for web search)

🛠️ Installation

1. Clone the repository

git clone https://github.com/tubone24/a2a_mastra.git
cd a2a_mastra

2. Copy the environment variables

cp .env.example .env

3. Configure your .env file

# AWS Credentials for Amazon Bedrock
AWS_ACCESS_KEY_ID=your-access-key-id
AWS_SECRET_ACCESS_KEY=your-secret-access-key
AWS_REGION=us-east-1

# Bedrock Model
BEDROCK_MODEL_ID=anthropic.claude-3-5-sonnet-20240620-v1:0

# Langfuse (optional)
LANGFUSE_PUBLIC_KEY=your-public-key
LANGFUSE_SECRET_KEY=your-secret-key
LANGFUSE_BASEURL=https://cloud.langfuse.com

# Brave Search (optional)
BRAVE_SEARCH_API_KEY=your-api-key

4. Build and start the services

docker-compose up --build

🎯 Usage

Once the system is running, access the frontend at http://localhost:3000.

Available Operations

1. Data Processing (/api/gateway/agents - type: process)

• Analyzes and transforms data
• Extracts patterns and insights

1. Summarization (/api/gateway/agents - type: summarize)

• Creates concise summaries
• Supports different audience types (technical, executive, general)

1. Analysis Workflow (/api/gateway/agents - type: analyze)

• Combines data processing and summarization
• End-to-end data analysis pipeline

1. Web Search (/api/gateway/agents - type: web-search)

• Real-time web information retrieval
• News and scholarly article search

1. Deep Research (/api/gateway/agents - type: deep-research)

• Multi-step research workflow using asynchronous task processing
• Combines web search, data processing, and summarization
• Long-running tasks with progress tracking and status polling

API Examples

# Analyze data with full workflow
curl -X POST http://localhost:3001/api/gateway/agents \
  -H "Content-Type: application/json" \
  -d '{
    "type": "analyze",
    "data": "Your data here",
    "options": {
      "audienceType": "executive"
    }
  }'

# Deep Research (Asynchronous)
curl -X POST http://localhost:3001/api/gateway/agents \
  -H "Content-Type: application/json" \
  -d '{
    "type": "deep-research",
    "topic": "AI trends in healthcare 2024",
    "options": {
      "depth": "comprehensive",
      "sources": ["web", "news", "academic"],
      "audienceType": "technical",
      "maxDuration": "10 minutes"
    }
  }'

# Response for Deep Research
{
  "taskId": "research-task-abc-123",
  "status": "initiated",
  "estimatedDuration": "8-10 minutes",
  "pollUrl": "/api/gateway/task/research-task-abc-123",
  "steps": {
    "total": 5,
    "current": 1,
    "phases": ["search", "analyze", "synthesize", "validate", "report"]
  }
}

# Poll for status
curl http://localhost:3001/api/gateway/task/research-task-abc-123

🔄 Communication Flows

Agent Discovery

The system implements a centralized agent discovery mechanism through the Gateway agent. The discovery process allows agents to register their capabilities and discover other agents in the network.

sequenceDiagram
    participant Frontend
    participant FrontendAPI
    participant Gateway
    participant DataProcessor
    participant Summarizer
    participant WebSearch
    
    Note over Frontend,WebSearch: Agent Discovery Initiation
    
    Frontend->>FrontendAPI: GET /api/gateway/agents
    activate FrontendAPI
    
    FrontendAPI->>Gateway: GET /api/gateway/agentshttp://gateway:3001
    activate Gateway
    
    Note over Gateway,WebSearch: Gateway Discovers All Agents
    
    Gateway->>DataProcessor: Mastra A2A.getCard()agentId: "data-processor-agent-01"Port: 3002
    activate DataProcessor
    DataProcessor-->>Gateway: Agent Card Response{id: "data-processor-agent-01", capabilities: ["data-analysis"], status: "online"}
    deactivate DataProcessor
    
    Gateway->>Summarizer: Mastra A2A.getCard()agentId: "summarizer-agent-01"Port: 3003
    activate Summarizer
    Summarizer-->>Gateway: Agent Card Response{id: "summarizer-agent-01", capabilities: ["text-summarization"], status: "online"}
    deactivate Summarizer
    
    Gateway->>WebSearch: HTTP GET /api/gateway/infoPort: 3004
    activate WebSearch
    WebSearch-->>Gateway: Agent Card Response{id: "web-search-agent-01", capabilities: ["web-search"], mcpEnabled: true}
    deactivate WebSearch
    
    Note over Gateway: Aggregate Agent Information
    
    Gateway-->>FrontendAPI: Discovery Response{gateway: {id: "gateway-agent-01", status: "online"},connectedAgents: [3 agents], totalAgents: 4}
    deactivate Gateway
    
    FrontendAPI-->>Frontend: Agent List Response{agents: [4 agents], discoveryTime: "150ms", onlineAgents: 4}
    deactivate FrontendAPI
    
    Note over Frontend: Display Agent Dashboard
    Frontend->>Frontend: Render Agent Cards- Status indicators- Capabilities- Supported task types

Gateway API Communication

The system implements REST API for Gateway and Mastra A2A protocol for other agents:

For Express Server Agent (Gateway):

1. REST API Discovery - HTTP endpoint /api/gateway/info for agent capability discovery
2. REST API Message Exchange - HTTP POST /api/gateway/message for synchronous communication
3. REST API Task Management - HTTP POST /api/gateway/task for asynchronous processing
4. REST API Task Streaming - HTTP GET /api/gateway/task/{id} for task progress polling

For Mastra Dev Server Agents (Data Processor, Summarizer, Web Search):

1. Agent Discovery - A2A.getAgentCard(agentId) - Agent capability discovery
2. Message Exchange - A2A.sendMessage({to, from, content}) - Synchronous communication
3. Task Management - A2A.createTask({agentId, taskType, payload}) - Asynchronous processing
4. Task Streaming - A2A.streamTaskUpdates(taskId) - Real-time task progress updates

Workflow Sequence with Agent Discovery

sequenceDiagram
    participant Client
    participant Gateway
    participant DataProcessor
    participant Summarizer
    
    Note over Gateway,Summarizer: Agent Discovery Phase (Mastra A2A)
    
    Gateway->>DataProcessor: A2A.getAgentCard("data-processor-agent-01")Port: 3002
    DataProcessor-->>Gateway: {agentId, name, capabilities,supportedTypes: ["process", "analyze"]}
    
    Gateway->>Summarizer: A2A.getAgentCard("summarizer-agent-01")Port: 3003
    Summarizer-->>Gateway: {agentId, name, capabilities,supportedTypes: ["summarize", "executive-summary"]}
    
    Note over Client,Summarizer: Workflow Execution Phase
    
    Client->>Gateway: POST /api/gateway/agents{type: "analyze", data: {...}}
    activate Gateway
    
    Gateway->>Gateway: Create Workflow ExecutionGenerate traceId & workflowId
    
    Gateway->>DataProcessor: A2A.sendMessage({  to: "data-processor-agent-01",  from: "gateway-agent-01",  content: {    type: "process",    data: {...},    metadata: {      workflowId: "wf-123",      traceId: "trace-456",      step: 1    }  }})
    activate DataProcessor
    
    DataProcessor->>DataProcessor: Process with BedrockTrack with Langfuse
    DataProcessor-->>Gateway: {  status: "success",  data: {processed_data, insights},  metadata: {processingTime: 1200ms}}
    deactivate DataProcessor
    
    Gateway->>Summarizer: A2A.sendMessage({  to: "summarizer-agent-01",  from: "gateway-agent-01",  content: {    type: "summarize",    data: processed_data,    options: {audienceType: "executive"},    metadata: {      workflowId: "wf-123",      traceId: "trace-456",      step: 2    }  }})
    activate Summarizer
    
    Summarizer->>Summarizer: Generate with BedrockTrack with Langfuse
    Summarizer-->>Gateway: {  status: "success",  data: {summary, keyPoints, recommendations},  metadata: {processingTime: 800ms}}
    deactivate Summarizer
    
    Gateway->>Gateway: Complete WorkflowAggregate Results
    Gateway-->>Client: {  workflowId: "wf-123",  status: "completed",  result: {processedData, summary},  totalDuration: 2000ms}
    deactivate Gateway

Web Search Flow with MCP Details

sequenceDiagram
    participant Client
    participant Gateway
    participant WebSearch
    participant MCPServer
    participant BraveAPI
    
    Note over Gateway,WebSearch: Agent Card Exchange via Mastra A2A
    
    Gateway->>WebSearch: A2A.getCard()agentId: "web-search-agent-01"Port: 3004
    WebSearch-->>Gateway: {  agentId: "web-search-agent-01",  name: "Web Search Agent",  capabilities: ["web-search", "news-search"],  mcpEnabled: true,  protocols: ["a2a/v1", "mcp/v1"]}
    
    Note over Client,BraveAPI: Search Request Flow
    
    Client->>Gateway: POST /api/gateway/agents{type: "web-search", query: "AI trends 2024"}
    activate Gateway
    
    Gateway->>WebSearch: A2A.sendMessage({  to: "web-search-agent-01",  from: "gateway-agent-01",  content: {    type: "search",    query: "AI trends 2024",    options: {limit: 10}  }})
    activate WebSearch
    
    Note over WebSearch,MCPServer: MCP Communication via Stdio
    
    WebSearch->>MCPServer: MCP Stdio Protocol{  method: "tool/call",  params: {    name: "brave_web_searc

…

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [tubone24](https://github.com/tubone24)
- **Source:** [tubone24/a2a_mastra](https://github.com/tubone24/a2a_mastra)
- **License:** MIT

Install and usage instructions live in the source repository linked above.