Multi-Agent Research System — A2A Protocol Showcase

System Architecture

Three specialized AI agents coordinated through a central orchestrator via the A2A protocol.

User Query Research UI / API

Orchestrator Task Planning & Routing

P

Perplexity Web Search & Sources

Real-time search Citations

C

Claude Analysis & Synthesis

Deep reasoning Writing

O

OpenAI GPT-4 Validation & QA

Cross-check Multimodal

Final Report PDF / DOCX / Markdown / JSON

1

Planning

Orchestrator parses the query, identifies sub-topics, and creates a research plan

2

Discovery

Perplexity searches the web for relevant sources and citations

3

Analysis

Claude performs deep analysis and synthesizes findings into structured output

4

Validation

OpenAI cross-validates claims, checks for inconsistencies, and verifies accuracy

5

Synthesis

Final report assembled with citations, confidence scores, and follow-up questions

Key Features

A2A Protocol

JSON-RPC 2.0 based agent-to-agent communication with discoverable agent cards and typed skill schemas.

Real-Time Streaming

Server-Sent Events deliver live progress updates as agents collaborate. Watch the research unfold in real time.

Provider Fallbacks

Configurable priority chains with automatic failover. Local Ollama integration for offline operation.

Multi-Format Export

Download research as PDF, DOCX, Markdown, HTML, or JSON. Server-side rendering with fpdf2 and python-docx.

Admin Panel

Web-based admin interface to configure agents, set provider priorities, and manage the system.

REST + JSON-RPC API

Full REST API for programmatic access plus standards-compliant A2A JSON-RPC endpoint for agent interop.

Interactive Demo

See the multi-agent workflow in action. This is a simulated demo — no API keys required.

Try:

P

Perplexity Web Search

C

Claude Analysis

O

OpenAI Validation

Research in Progress

0%

Initializing agents...

Agent Communication

Research Results

Simulated

Code Showcase

Key implementation details from the codebase.

src/a2a/protocol.py
class TaskStatus(str, Enum):
    """Task execution status states."""
    PENDING     = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED   = "completed"
    FAILED      = "failed"
    CANCELLED   = "cancelled"
    WAITING     = "waiting"


class AgentSkill(BaseModel):
    """Definition of a capability that an agent can perform."""
    id: str           = Field(..., description="Unique skill identifier")
    name: str         = Field(..., description="Human-readable name")
    description: str  = Field(..., description="What this skill does")
    input_schema: Optional[dict]  = Field(None, alias="inputSchema")
    output_schema: Optional[dict] = Field(None, alias="outputSchema")
    tags: list[str]   = Field(default_factory=list)


class AgentCard(BaseModel):
    """Discoverable manifest describing an agent's identity and skills."""
    name: str
    description: str
    url: Optional[HttpUrl]   = None
    version: str              = "1.0.0"
    skills: list[AgentSkill] = Field(default_factory=list)
    capabilities: AgentCapabilities = Field(default_factory=AgentCapabilities)
src/agents/base.py
class BaseAgent(ABC):
    """Abstract base class for all A2A agents.

    Each agent implementation must:
    1. Provide an agent card describing its capabilities
    2. Implement the execute method for processing tasks
    3. Optionally support streaming responses
    """

    def __init__(self, agent_id: str, api_key: Optional[str] = None):
        self.agent_id = agent_id
        self.api_key = api_key
        self._initialized = False

    @abstractmethod
    async def execute(self, task: Task) -> TaskResponse:
        """Execute a task and return the result."""
        ...

    @abstractmethod
    async def execute_streaming(self, task: Task) -> AsyncIterator[TaskResponse]:
        """Execute with streaming intermediate results."""
        ...


@dataclass
class ProviderPriority:
    """Priority config for a skill's provider fallback chain."""
    skill_id: str
    primary: str
    fallbacks: list[str] = field(default_factory=list)

    def get_ordered_agents(self) -> list[str]:
        return [self.primary] + self.fallbacks
src/workflows/research.py
class ResearchDepth(str, Enum):
    QUICK    = "quick"       # 1-2 searches, basic analysis
    STANDARD = "standard"    # 3-4 searches, full analysis
    DEEP     = "deep"        # 5+ searches, extensive validation


@dataclass
class ResearchConfig:
    depth: ResearchDepth        = ResearchDepth.STANDARD
    output_format: OutputFormat = OutputFormat.REPORT
    max_sources: int            = 10
    include_validation: bool    = True
    timeout_seconds: int        = 300


@dataclass
class ResearchResult:
    """Result of a research workflow execution."""
    query: str
    report: str
    sources: list[dict]
    artifacts: list[Artifact]
    duration_seconds: float
    success: bool
    error: Optional[str] = None

    def to_dict(self) -> dict:
        return {
            "query": self.query,
            "report": self.report,
            "sources_count": len(self.sources),
            "duration_seconds": self.duration_seconds,
            "success": self.success,
        }
src/main.py
@app.post("/research")
async def execute_research(request: ResearchRequest):
    """Execute a research workflow."""
    config = ResearchConfig(
        depth=ResearchDepth(request.depth),
        output_format=OutputFormat(request.output_format)
    )

    if request.stream:
        return EventSourceResponse(
            stream_research(request.query, config),
            media_type="text/event-stream"
        )

    result = await research_workflow.execute(request.query, config)

    return {
        "success": result.success,
        "query": result.query,
        "report": result.report,
        "sources_count": len(result.sources),
        "duration_seconds": result.duration_seconds,
    }


@app.post("/a2a")
async def a2a_endpoint(request: A2ARequest):
    """A2A Protocol JSON-RPC endpoint."""
    valid, error = A2AProtocol.validate_request(request.model_dump())
    if not valid:
        return A2AProtocol.create_error(
            request.id, A2AProtocol.ERROR_INVALID_REQUEST, error
        )

    if method == A2AProtocol.METHOD_TASK_SEND:
        return await handle_task_send(request.id, params)
    elif method == A2AProtocol.METHOD_TASK_GET:
        return await handle_task_get(request.id, params)
A2A Protocol — JSON-RPC Message Flow
// 1. Orchestrator sends task to Perplexity for web search
{
    "jsonrpc": "2.0",
    "method": "tasks/send",
    "params": {
        "id": "task-001",
        "message": {
            "role": "user",
            "parts": [{
                "type": "text",
                "text": "Search for AI healthcare FDA approvals 2024"
            }]
        },
        "skillId": "web-search",
        "agentId": "perplexity"
    },
    "id": "req-001"
}

// 2. Perplexity returns results with sources
{
    "jsonrpc": "2.0",
    "result": {
        "id": "task-001",
        "status": "completed",
        "artifacts": [{
            "type": "application/json",
            "parts": [{"type": "data", "data": {"sources": [...]}}]
        }]
    },
    "id": "req-001"
}

API Reference

Endpoint	Method	Description
`/research`	POST	Execute a multi-agent research workflow
`/a2a`	POST	A2A Protocol JSON-RPC endpoint
`/agents`	GET	List all registered agents
`/agents/{id}/card`	GET	Get agent card (A2A discovery)
`/export/pdf`	POST	Download report as PDF
`/export/docx`	POST	Download report as DOCX
`/health`	GET	System health check
`/api/info`	GET	System information

Technical Stack

Backend

Python 3.12 + asyncio
FastAPI + Pydantic
SSE via sse-starlette
Uvicorn ASGI server

AI Providers

Anthropic Claude (Analysis)
OpenAI GPT-4 (Validation)
Perplexity (Web Search)
Ollama (Local Fallback)

Protocol

A2A Agent-to-Agent
JSON-RPC 2.0
Agent Cards (Discovery)
Typed Skill Schemas

Frontend & Export

Vanilla JS, CSS3
marked.js (Markdown)
fpdf2 (PDF generation)
python-docx (DOCX)