MCP and A2A protocol integration (v1.10.5)

- MCP client (stdio/HTTP/SSE), MCPToolAdapter, discovery and config
- A2A client, server, A2AAgentTool, discovery and config
- CLI: hivemind mcp list|test|add, hivemind a2a serve|discover|call
- Doctor checks for MCP servers and A2A agents
- Protocol version constants in hivemind/protocols/
- Optional deps: mcp, a2a in pyproject.toml
- Tests: tests/test_protocols.py

Made-with: Cursor

Author: rithulkamesh

CHANGELOG.md

@@ -7,6 +7,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.10.5] - 2026-03-10
+
+### Added
+
+- MCP (Model Context Protocol) client supporting stdio, HTTP, and SSE transports
+- MCPToolAdapter: MCP server tools appear in hivemind tool registry automatically
+- MCP server config via `[[mcp.servers]]` in hivemind.toml
+- A2A (Agent-to-Agent) client for calling external A2A-compliant agents as tools
+- A2AServer: exposes hivemind as an A2A endpoint with AgentCard and task endpoints
+- A2AAgentTool: external A2A agent skills registered as local tools
+- `hivemind mcp list|test|add` CLI commands
+- `hivemind a2a serve|discover|call` CLI commands
+- Protocol version constants in `hivemind/protocols/`
+
+### Changed
+
+- Tool category "mcp" and "a2a" added to scoring and selection pipeline
+
 ## [1.10.0] - 2026-03-10
 
 ### Added

hivemind/agents/a2a/__init__.py

@@ -0,0 +1,27 @@
+"""
+A2A (Agent-to-Agent) integration: client, server, tool adapter.
+"""
+
+from hivemind.agents.a2a.types import (
+    AgentCard,
+    AgentSkill,
+    A2ATaskRequest,
+    A2ATaskResponse,
+)
+from hivemind.agents.a2a.client import A2AClient
+from hivemind.agents.a2a.server import create_a2a_app, run_a2a_server
+from hivemind.agents.a2a.tool_adapter import A2AAgentTool
+
+# Alias for task spec: "A2AServer" = run_a2a_server / create_a2a_app
+A2AServer = create_a2a_app
+
+__all__ = [
+    "A2AClient",
+    "A2AServer",
+    "AgentCard",
+    "A2ATaskRequest",
+    "A2ATaskResponse",
+    "A2AAgentTool",
+    "create_a2a_app",
+    "run_a2a_server",
+]

hivemind/agents/a2a/client.py

@@ -0,0 +1,126 @@
+"""
+A2A client: call external A2A-compliant agents (get AgentCard, send_task, stream_task).
+"""
+
+import uuid
+from typing import AsyncIterator
+
+from hivemind.agents.a2a.types import (
+    AgentCard,
+    AgentSkill,
+    A2ATaskRequest,
+    A2ATaskResponse,
+)
+
+
+def _parse_agent_card(data: dict) -> AgentCard:
+    """Build AgentCard from JSON response."""
+    skills_data = data.get("skills") or []
+    skills = [
+        AgentSkill(
+            id=s.get("id", ""),
+            name=s.get("name", ""),
+            description=s.get("description", ""),
+            input_modes=list(s.get("inputModes") or []),
+            output_modes=list(s.get("outputModes") or []),
+        )
+        for s in skills_data
+        if isinstance(s, dict)
+    ]
+    return AgentCard(
+        name=data.get("name", ""),
+        description=data.get("description", ""),
+        url=data.get("url", ""),
+        version=data.get("version", ""),
+        capabilities=list(data.get("capabilities") or []),
+        skills=skills,
+        authentication=data.get("authentication"),
+    )
+
+
+def _parse_task_response(data: dict) -> A2ATaskResponse:
+    """Build A2ATaskResponse from JSON."""
+    status = data.get("status", "failed")
+    if status not in ("submitted", "working", "completed", "failed", "canceled"):
+        status = "failed"
+    return A2ATaskResponse(
+        id=data.get("id", ""),
+        status=status,
+        result=data.get("result"),
+        artifacts=data.get("artifacts") or [],
+    )
+
+
+class A2AClient:
+    """Calls external A2A-compliant agents as if they were local tools."""
+
+    def __init__(self, timeout_seconds: float = 60.0) -> None:
+        self.timeout_seconds = timeout_seconds
+
+    def _client(self):
+        import httpx
+        return httpx.AsyncClient(timeout=self.timeout_seconds)
+
+    async def get_agent_card(self, base_url: str) -> AgentCard:
+        """GET {base_url}/.well-known/agent.json"""
+        url = base_url.rstrip("/") + "/.well-known/agent.json"
+        async with self._client() as client:
+            r = await client.get(url)
+            r.raise_for_status()
+            data = r.json()
+        return _parse_agent_card(data)
+
+    async def send_task(
+        self,
+        base_url: str,
+        request: A2ATaskRequest,
+        poll_interval: float = 0.5,
+    ) -> A2ATaskResponse:
+        """POST {base_url}/tasks/send, then poll GET /tasks/{id} until completed/failed."""
+        import asyncio
+        url = base_url.rstrip("/") + "/tasks/send"
+        body = {
+            "id": request.id,
+            "message": request.message,
+            "sessionId": request.session_id,
+        }
+        async with self._client() as client:
+            r = await client.post(url, json=body)
+            r.raise_for_status()
+            data = r.json()
+        resp = _parse_task_response(data)
+        if resp.status in ("completed", "failed", "canceled"):
+            return resp
+        task_url = base_url.rstrip("/") + f"/tasks/{resp.id}"
+        while resp.status in ("submitted", "working"):
+            await asyncio.sleep(poll_interval)
+            async with self._client() as client:
+                r = await client.get(task_url)
+                r.raise_for_status()
+                data = r.json()
+            resp = _parse_task_response(data)
+            if resp.status in ("completed", "failed", "canceled"):
+                return resp
+        return resp
+
+    async def stream_task(
+        self,
+        base_url: str,
+        request: A2ATaskRequest,
+    ) -> AsyncIterator[str]:
+        """POST {base_url}/tasks/sendSubscribe, consume SSE stream, yield text chunks."""
+        import httpx
+        url = base_url.rstrip("/") + "/tasks/sendSubscribe"
+        body = {
+            "id": request.id,
+            "message": request.message,
+            "sessionId": request.session_id,
+        }
+        async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            async with client.stream("POST", url, json=body) as response:
+                response.raise_for_status()
+                async for line in response.aiter_lines():
+                    if line.startswith("data:"):
+                        data = line[5:].strip()
+                        if data and data != "[DONE]":
+                            yield data

hivemind/agents/a2a/discovery.py

@@ -0,0 +1,24 @@
+"""
+A2A discovery: fetch AgentCard from URL, register skills as A2AAgentTool.
+"""
+
+import asyncio
+
+from hivemind.config.schema import A2AAgentConfig
+from hivemind.agents.a2a.client import A2AClient
+from hivemind.agents.a2a.tool_adapter import A2AAgentTool
+
+
+async def discover_a2a_tools_async(agent_config: A2AAgentConfig) -> int:
+    """Fetch AgentCard from agent URL, create A2AAgentTool per skill, register. Return count."""
+    client = A2AClient()
+    card = await client.get_agent_card(agent_config.url)
+    name = agent_config.name or card.name or "a2a-agent"
+    for skill in card.skills:
+        A2AAgentTool(agent_name=name, skill=skill, base_url=agent_config.url, client=client)
+    return len(card.skills)
+
+
+def register_a2a_agent(agent_config: A2AAgentConfig) -> int:
+    """Sync: discover and register A2A agent tools; return count."""
+    return asyncio.run(discover_a2a_tools_async(agent_config))

hivemind/agents/a2a/server.py

@@ -0,0 +1,128 @@
+"""
+A2A server: expose hivemind agents as A2A-compliant endpoints (FastAPI).
+"""
+
+from typing import Any
+
+from hivemind.agents.a2a.types import AgentCard, AgentSkill, A2ATaskRequest, A2ATaskResponse
+
+
+def _build_agent_card(host: str, port: int, swarm_name: str) -> dict:
+    """Build AgentCard JSON for GET /.well-known/agent.json."""
+    from hivemind.intelligence.strategy_selector import ExecutionStrategy
+    url = f"http://{host}:{port}"
+    name = swarm_name or "hivemind"
+    skills = [
+        {
+            "id": s.value,
+            "name": s.value.replace("_", " ").title(),
+            "description": f"Execute task using {s.value} strategy.",
+            "inputModes": ["text"],
+            "outputModes": ["text"],
+        }
+        for s in ExecutionStrategy
+        if s != ExecutionStrategy.GENERAL
+    ]
+    return {
+        "name": name,
+        "description": "Hivemind swarm orchestration as A2A agent.",
+        "url": url,
+        "version": "1.10.5",
+        "capabilities": ["streaming"],
+        "skills": skills,
+    }
+
+
+def create_a2a_app(host: str = "localhost", port: int = 8080, swarm_name: str = "") -> Any:
+    """Create FastAPI app with A2A routes. Requires fastapi, uvicorn, sse-starlette."""
+    from fastapi import FastAPI
+    from fastapi.responses import StreamingResponse
+    import asyncio
+    import uuid
+
+    app = FastAPI(title="Hivemind A2A Server")
+
+    @app.get("/.well-known/agent.json")
+    def agent_card() -> dict:
+        return _build_agent_card(host, port, swarm_name)
+
+    @app.post("/tasks/send")
+    async def tasks_send(body: dict) -> dict:
+        """Accept A2A task, run via Swarm, return result."""
+        task_id = body.get("id") or str(uuid.uuid4())
+        message = body.get("message") or {}
+        text = message.get("text", "") if isinstance(message, dict) else ""
+        if not text:
+            return {
+                "id": task_id,
+                "status": "failed",
+                "result": "Missing message.text",
+                "artifacts": [],
+            }
+        try:
+            from hivemind.config import get_config
+            from hivemind.swarm.swarm import Swarm
+            cfg = get_config()
+            swarm = Swarm(config=cfg)
+            result = swarm.run(text)
+            out = "\n".join(f"{k}: {v[:2000]}" for k, v in result.items()) if result else ""
+            return {
+                "id": task_id,
+                "status": "completed",
+                "result": out,
+                "artifacts": [],
+            }
+        except Exception as e:
+            return {
+                "id": task_id,
+                "status": "failed",
+                "result": str(e),
+                "artifacts": [],
+            }
+
+    @app.post("/tasks/sendSubscribe")
+    async def tasks_send_subscribe(body: dict):
+        """SSE streaming task execution."""
+        task_id = body.get("id") or str(uuid.uuid4())
+        message = body.get("message") or {}
+        text = message.get("text", "") if isinstance(message, dict) else ""
+
+        async def stream():
+            if not text:
+                yield f"data: {__import__('json').dumps({'error': 'Missing message.text'})}\n\n"
+                return
+            try:
+                from hivemind.config import get_config
+                from hivemind.swarm.swarm import Swarm
+                cfg = get_config()
+                swarm = Swarm(config=cfg)
+                result = swarm.run(text)
+                out = "\n".join(f"{k}: {v[:2000]}" for k, v in result.items()) if result else ""
+                yield f"data: {__import__('json').dumps({'id': task_id, 'status': 'completed', 'result': out})}\n\n"
+            except Exception as e:
+                yield f"data: {__import__('json').dumps({'id': task_id, 'status': 'failed', 'result': str(e)})}\n\n"
+
+        return StreamingResponse(
+            stream(),
+            media_type="text/event-stream",
+            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
+        )
+
+    @app.get("/tasks/{task_id}")
+    def task_status(task_id: str) -> dict:
+        """Task status (stub: we don't persist task state for now)."""
+        return {"id": task_id, "status": "completed", "result": None, "artifacts": []}
+
+    @app.post("/tasks/{task_id}/cancel")
+    def task_cancel(task_id: str) -> dict:
+        """Cancel running task (stub)."""
+        return {"id": task_id, "status": "canceled", "result": None, "artifacts": []}
+
+    return app
+
+
+def run_a2a_server(host: str = "localhost", port: int = 8080, swarm_name: str = "") -> None:
+    """Run A2A server with uvicorn."""
+    import uvicorn
+    app = create_a2a_app(host=host, port=port, swarm_name=swarm_name)
+    uvicorn.run(app, host=host, port=port)