A2A Integrations

Connect any external service to a Multica agent over A2A v1.0. Recipes for Python (a2a-sdk), curl, LangChain, CrewAI, n8n, Claude Code, and a generic JSON-RPC consumer.

Once an agent is exposed via A2A (Settings → A2A on the agent — see A2A Protocol), any service that speaks A2A v1.0 can call it like a remote function. This page is the cookbook.

What you get

Pointing an external system at a Multica A2A agent gives you, in one stable API:

Synchronous task execution — fire-and-wait or fire-and-poll, your choice via returnImmediately.
Streaming progress — SSE feed of state transitions and artifacts as the agent works.
Push notifications — agent pings your webhook on every status change.
Multi-agent coordination — one orchestrator can address many agents, even across workspaces, by switching the URL + token.
No coupling to Multica's UI or daemon — your service talks pure A2A; if Multica changes its internal task pipeline tomorrow, your integration doesn't notice.

This is genuinely different from "call our REST API and parse the response" — A2A standardises the contract (Agent Card → Task → Artifacts), so any A2A-aware client works with any A2A-aware agent. You write the integration once and reuse it against other A2A-published agents too.

Prerequisites

For every recipe below you need three things from Multica:

BASE_URL   = https://a2a.magomadov.dev/agents/<agent-id>     # see Settings → A2A → Agent Card preview
TOKEN      = mca_…                                            # see Settings → A2A → Mint token
HEADERS    = Authorization: Bearer <TOKEN>
             A2A-Version: 1.0

Find <agent-id> in the URL when you open an agent in Multica (/<workspace>/agents/<id>), or via zagents agent list.

Recipe 1 — Python with the official SDK

The fastest way to integrate is the a2a-sdk. Handles JSON-RPC, REST, SSE, push notifications, schema validation — you write await client.send_message(...).

pip install "a2a-sdk[http-server]"

import asyncio
import httpx
from a2a.client import A2AClient
from a2a.types.a2a_pb2 import Message, Part, Role, SendMessageConfiguration
import uuid

BASE_URL = "https://a2a.magomadov.dev/agents/<agent-id>"
TOKEN = "mca_…"

async def main() -> None:
    headers = {"Authorization": f"Bearer {TOKEN}", "A2A-Version": "1.0"}
    async with httpx.AsyncClient(headers=headers, timeout=120) as http:
        client = A2AClient(httpx_client=http, agent_card_url=f"{BASE_URL}/.well-known/agent-card.json")

        msg = Message(message_id=str(uuid.uuid4()), role=Role.ROLE_USER)
        msg.parts.append(Part(text="Summarize the latest open PRs in repo X."))

        # Blocking — wait until the agent finishes (default).
        task = await client.send_message(message=msg)
        print(f"state: {task.status.state}")          # TASK_STATE_COMPLETED
        for art in task.artifacts:
            for part in art.parts:
                if part.HasField("text"):
                    print(part.text)

asyncio.run(main())

Streaming variant:

async for ev in client.send_streaming_message(message=msg):
    if ev.HasField("status_update"):
        print("status →", ev.status_update.status.state)
    elif ev.HasField("artifact_update"):
        print("artifact →", ev.artifact_update.artifact.name)

Recipe 2 — curl (no SDK, debugging or shell scripts)

# 1. Send a message, get the task back immediately
curl -s -X POST "$BASE_URL/message:send" \
  -H "Authorization: Bearer $TOKEN" \
  -H "A2A-Version: 1.0" \
  -H "Content-Type: application/json" \
  -d '{
    "message": {
      "messageId": "'$(uuidgen)'",
      "role": "ROLE_USER",
      "parts": [{"text": "ping"}]
    },
    "configuration": {"returnImmediately": true}
  }' | tee /tmp/task.json

# 2. Poll until terminal
TASK_ID=$(jq -r .task.id /tmp/task.json)
while :; do
  STATE=$(curl -s -H "Authorization: Bearer $TOKEN" -H "A2A-Version: 1.0" \
    "$BASE_URL/tasks/$TASK_ID" | jq -r .status.state)
  echo "$STATE"
  case "$STATE" in
    TASK_STATE_COMPLETED|TASK_STATE_FAILED|TASK_STATE_CANCELED|TASK_STATE_REJECTED) break ;;
  esac
  sleep 1
done

SSE streaming:

curl -N -X POST "$BASE_URL/message:stream" \
  -H "Authorization: Bearer $TOKEN" -H "A2A-Version: 1.0" \
  -H "Content-Type: application/json" \
  -d '{"message":{"messageId":"'$(uuidgen)'","role":"ROLE_USER","parts":[{"text":"stream me"}]}}'

Recipe 3 — LangChain

A Multica A2A agent becomes a LangChain Tool that any agent / chain can call.

pip install langchain "a2a-sdk[http-server]"

from langchain.tools import StructuredTool
import asyncio, httpx, uuid
from a2a.client import A2AClient
from a2a.types.a2a_pb2 import Message, Part, Role

def make_multica_tool(agent_id: str, token: str, name: str, description: str) -> StructuredTool:
    base_url = f"https://a2a.magomadov.dev/agents/{agent_id}"

    async def _call(prompt: str) -> str:
        headers = {"Authorization": f"Bearer {token}", "A2A-Version": "1.0"}
        async with httpx.AsyncClient(headers=headers, timeout=300) as http:
            client = A2AClient(
                httpx_client=http,
                agent_card_url=f"{base_url}/.well-known/agent-card.json",
            )
            msg = Message(message_id=str(uuid.uuid4()), role=Role.ROLE_USER)
            msg.parts.append(Part(text=prompt))
            task = await client.send_message(message=msg)
            return "\n".join(
                p.text for art in task.artifacts for p in art.parts if p.HasField("text")
            )

    def _sync_call(prompt: str) -> str:
        return asyncio.run(_call(prompt))

    return StructuredTool.from_function(
        name=name,
        description=description,
        func=_sync_call,
    )

# Now you can drop it into any LangChain agent
multica_search = make_multica_tool(
    agent_id="<agent-id>", token="mca_…",
    name="multica_search",
    description="Searches our internal knowledge base. Pass a natural-language query.",
)

Recipe 4 — CrewAI

CrewAI has a similar tool abstraction.

from crewai_tools import BaseTool
import asyncio, httpx, uuid
from a2a.client import A2AClient
from a2a.types.a2a_pb2 import Message, Part, Role

class MulticaA2ATool(BaseTool):
    name: str = "multica_agent"
    description: str = "Calls the Multica research agent over A2A."

    def __init__(self, agent_id: str, token: str, **kwargs):
        super().__init__(**kwargs)
        self._base = f"https://a2a.magomadov.dev/agents/{agent_id}"
        self._headers = {"Authorization": f"Bearer {token}", "A2A-Version": "1.0"}

    def _run(self, prompt: str) -> str:
        async def go():
            async with httpx.AsyncClient(headers=self._headers, timeout=300) as http:
                c = A2AClient(httpx_client=http, agent_card_url=f"{self._base}/.well-known/agent-card.json")
                m = Message(message_id=str(uuid.uuid4()), role=Role.ROLE_USER)
                m.parts.append(Part(text=prompt))
                task = await c.send_message(message=m)
                return "\n".join(p.text for a in task.artifacts for p in a.parts if p.HasField("text"))
        return asyncio.run(go())

Plug it into a Crew:

from crewai import Agent, Task, Crew

researcher = Agent(role="Researcher", goal="answer questions",
                   tools=[MulticaA2ATool(agent_id="…", token="mca_…")])
crew = Crew(agents=[researcher], tasks=[Task(description="Find …", agent=researcher)])
crew.kickoff()

Recipe 5 — n8n / Make / Zapier (no-code)

A2A is plain HTTPS + JSON, so the HTTP Request node in any no-code platform calls it directly. Configure once:

Node setting	Value
Method	`POST`
URL	`{{$env.MULTICA_A2A_BASE}}/message:send`
Headers	`Authorization: Bearer {{$env.MULTICA_A2A_TOKEN}}`, `A2A-Version: 1.0`
Body (JSON)	`{ "message": { "messageId": "{{$workflow.id}}-{{$itemIndex}}", "role": "ROLE_USER", "parts": [{"text": "{{$json.prompt}}"}] }, "configuration": { "returnImmediately": false } }`

The response contains the completed Task with artifacts; map {{$json.task.artifacts[0].parts[0].text}} into the next node.

Recipe 6 — Claude Code (or any MCP-aware coding tool)

Wrap the Multica agent as an MCP tool. A minimal FastAPI MCP shim:

# multica_mcp.py — MCP server exposing one tool that calls Multica
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx, uuid

mcp = Server("multica-a2a-bridge")
TOKEN = "mca_…"
BASE = "https://a2a.magomadov.dev/agents/<agent-id>"

@mcp.list_tools()
async def list_tools():
    return [Tool(name="ask_multica_agent", description="Ask the Multica research agent",
                 inputSchema={"type": "object", "properties": {"prompt": {"type": "string"}}, "required": ["prompt"]})]

@mcp.call_tool()
async def call_tool(name: str, arguments: dict):
    if name != "ask_multica_agent":
        raise ValueError("unknown tool")
    body = {
        "message": {"messageId": str(uuid.uuid4()), "role": "ROLE_USER",
                    "parts": [{"text": arguments["prompt"]}]},
        "configuration": {"returnImmediately": False},
    }
    async with httpx.AsyncClient(headers={"Authorization": f"Bearer {TOKEN}", "A2A-Version": "1.0"}, timeout=300) as h:
        r = await h.post(f"{BASE}/message:send", json=body)
        task = r.json()["task"]
    text = "\n".join(p["text"] for a in task.get("artifacts", []) for p in a["parts"] if "text" in p)
    return [TextContent(type="text", text=text)]

Add to Claude Code's MCP config, restart, and ask_multica_agent shows up as a callable tool inside Claude Code sessions.

Recipe 7 — Generic JSON-RPC consumer (any language)

Every operation is also reachable through the JSON-RPC binding at BASE_URL/jsonrpc:

curl -s -X POST "$BASE_URL/jsonrpc" \
  -H "Authorization: Bearer $TOKEN" -H "A2A-Version: 1.0" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "1",
    "method": "SendMessage",
    "params": {
      "message": {"messageId": "'$(uuidgen)'", "role": "ROLE_USER", "parts": [{"text": "via jsonrpc"}]},
      "configuration": {"returnImmediately": true}
    }
  }' | jq .

Useful when your stack already has a JSON-RPC client (Go's gorilla/rpc, Rust's jsonrpsee, etc.) — same endpoint, every method (SendMessage, GetTask, CancelTask, SubscribeToTask, CreateTaskPushNotificationConfig, …) wrapped in JSON-RPC envelopes.

Push notifications — fire-and-forget integrations

Useful when your service can't keep an HTTP connection open. Register your webhook once per task:

# After creating a task with returnImmediately=true and capturing $TASK_ID:
curl -s -X POST "$BASE_URL/tasks/$TASK_ID/pushNotificationConfigs" \
  -H "Authorization: Bearer $TOKEN" -H "A2A-Version: 1.0" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-service.example.com/webhooks/multica",
    "token": "your-shared-verification-token"
  }'

The sidecar will POST a StreamResponse-shaped body to your webhook on every status change and artifact, with the token echoed back so you can verify the call originated from Multica. Configs are auto-deleted when the task reaches a terminal state.

Multi-agent / multi-workspace orchestration

A common pattern: one external orchestrator addresses several Multica agents.

agents = {
    "research":  ("https://a2a.magomadov.dev/agents/<research-id>",  "mca_research_…"),
    "code":      ("https://a2a.magomadov.dev/agents/<code-id>",      "mca_code_…"),
    "summarize": ("https://a2a.magomadov.dev/agents/<sum-id>",       "mca_sum_…"),
}

async def call(agent: str, prompt: str) -> str:
    base, token = agents[agent]
    headers = {"Authorization": f"Bearer {token}", "A2A-Version": "1.0"}
    async with httpx.AsyncClient(headers=headers, timeout=300) as http:
        c = A2AClient(httpx_client=http, agent_card_url=f"{base}/.well-known/agent-card.json")
        m = Message(message_id=str(uuid.uuid4()), role=Role.ROLE_USER)
        m.parts.append(Part(text=prompt))
        task = await c.send_message(message=m)
        return "\n".join(p.text for a in task.artifacts for p in a.parts if p.HasField("text"))

facts = await call("research", "what changed in the React 19 release?")
patch = await call("code", f"Apply these changes:\n{facts}")
note  = await call("summarize", patch)

Each agent has its own token (mint one per consumer in Settings → A2A → Mint token — easier to revoke individually) and the same A2A-version contract.

Operational tips

Sessions / context — pass the same contextId on follow-up messages to keep the conversation in one Multica chat_session. Different contextIds run in parallel; same contextId serializes.
Quotas — defaults: 60 req/min per token, 300 burst, 50 concurrent A2A tasks per workspace, agent's MaxConcurrentTasks (default 6) per agent. Quota excess returns google.rpc.Status with reason: "WORKSPACE_QUOTA_EXCEEDED" or RESOURCE_EXHAUSTED.
Audit — every external call lands in Settings → A2A → Recent A2A activity so you can see which agent / status / when. The full row is in agent_task_queue with source='a2a'.
Validate — point a2a-inspector at any Multica agent's card URL to spot-check the contract (full procedure in A2A Architecture).

Reference

A2A is a published spec, not a Multica-specific API. The same client code works against any A2A-published agent — Google's published agents, third-party agents, your own non-Multica deployments. Build the integration once, reuse it everywhere.

On this page