JSON-RPC is a lightweight, stateless remote procedure call (RPC) protocol encoded in JSON, often used for communication between client and server applications. Below is an explanation and a basic example of using JSON-RPC in Python.

What is JSON-RPC?

• JSON-RPC sends requests as JSON objects describing the method to call, its parameters, and an ID for tracking the response.
• The server responds with a JSON object containing either the result or an error, along with the same ID for correlation.
• It is transport-agnostic—can run over HTTP, WebSocket, etc.—and is commonly found in blockchain and API integrations.

Example: JSON-RPC in Python

Server Example

The following Python code creates a simple JSON-RPC server using the json-rpc library and Werkzeug:

from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple
from jsonrpc import JSONRPCResponseManager, dispatcher

@dispatcher.add_method
def foobar(**kwargs):
    return kwargs["foo"] + kwargs["bar"]

@Request.application
def application(request):
    dispatcher["echo"] = lambda s: s
    dispatcher["add"] = lambda a, b: a + b

    response = JSONRPCResponseManager.handle(
        request.data, dispatcher)
    return Response(response.json, mimetype='application/json')

if __name__ == '__main__':
    run_simple('localhost', 4000, application)

This server can handle "add", "echo", and "foobar" methods via JSON-RPC.

Client Example

A simple client using the requests library:

import requests
import json

def main():
    url = "http://localhost:4000/jsonrpc"
    headers = {'content-type': 'application/json'}
    payload = {
        "method": "echo",
        "params": ["echome!"],
        "jsonrpc": "2.0",
        "id": 0,
    }
    response = requests.post(url, data=json.dumps(payload), headers=headers).json()
    print(response)

if __name__ == "__main__":
    main()

This client sends an "echo" call and prints the server's response.

Typical JSON-RPC Message Structure

• Request:

  {
    "jsonrpc": "2.0",
    "method": "add",
    "params": [3, 4],
    "id": 1
  }

• Response:

  {
    "jsonrpc": "2.0",
    "result": 7,
    "id": 1
  }

The server executes the requested method and returns the result in this format.

JSON-RPC serves as the foundational communication layer for multiple AI agent protocols, enabling standardized remote procedure calls that facilitate seamless interaction between autonomous AI systems. The Agent2Agent (A2A) Protocol specifically leverages JSON-RPC 2.0 to enable AI agents to communicate, collaborate, and coordinate tasks across different platforms and vendors.

JSON-RPC as the Communication Foundation

JSON-RPC 2.0 is a lightweight, stateless remote procedure call protocol that uses JSON as the data format. In the context of AI agents, it provides:

• Standardized message structure with method, params, and id fields for request correlation
• Language-agnostic communication that works across different AI frameworks and platforms
• Transport flexibility over HTTP, WebSockets, or other protocols

The Agent2Agent (A2A) Protocol

A2A is an open standard designed to facilitate communication and interoperability between independent AI agent systems. Originally developed by Google and now governed by the Linux Foundation, A2A addresses the critical challenge of enabling AI agents built on diverse frameworks to work together effectively.

Core Architecture

A2A operates on a client-remote agent communication model where:

• Client agents initiate tasks and send requests to specialized remote agents
• Remote agents process tasks and return results or complete specific actions
• Agents maintain independence without sharing memory or tools by default
• Communication occurs through structured JSON-RPC messages over HTTPS

JSON-RPC Implementation in A2A

A2A uses JSON-RPC 2.0 as the message exchange mechanism. The protocol structure includes:

{
  "jsonrpc": "2.0",
  "method": "message/send",
  "params": {
    "task_id": "task-123",
    "message": {
      "role": "user",
      "parts": [
        {
          "type": "text",
          "content": "Optimize inventory levels for predicted demand spike"
        }
      ]
    }
  },
  "id": 1
}

Messages contain structured "parts" that can include different formats like text, images, or audio, enabling flexible multimodal interactions.

AI Agent Communication Workflow

The typical A2A communication flow demonstrates how JSON-RPC enables agent coordination:

Discovery Phase

Agents publish Agent Cards (JSON metadata documents) at well-known URLs that describe their capabilities, supported tasks, and endpoint details.

Authentication & Authorization

Client agents authenticate using OpenAPI-compatible schemes like OAuth 2.0 or API keys before establishing communication.

Task Execution

1. Task Initiation: Client sends JSON-RPC request with task parameters
2. Processing: Remote agent processes the request and may send progress updates via Server-Sent Events (SSE)
3. Response: Agent returns results or artifacts through JSON-RPC response format

Long-Running Operations

For complex tasks requiring extended processing time, A2A supports task objects that enable asynchronous coordination:

{
  "jsonrpc": "2.0",
  "result": {
    "task_id": "supply-chain-optimization-456",
    "status": "in_progress"
  },
  "id": 1
}

Comparison with Other AI Agent Protocols

A2A differs from other emerging protocols in its focus and implementation approach:

Enterprise Implementation Benefits

A2A's JSON-RPC foundation provides several enterprise advantages:

• Standards-based integration using familiar HTTP and JSON technologies
• Enterprise-grade security with established authentication mechanisms
• Scalable architecture supporting both synchronous and asynchronous operations
• Vendor neutrality enabling agents from different providers to collaborate
• Transport flexibility working over existing network infrastructure

Python Implementation Example

A basic A2A server implementation using the specialized a2a-json-rpc library:

import asyncio
from a2a_json_rpc.protocol import JSONRPCProtocol
from a2a_json_rpc.models import Json

# Create A2A-specific protocol instance
protocol = JSONRPCProtocol()

# Register agent method handler
@protocol.method("task/process")
async def process_task(method: str, params: Json) -> Json:
    task_id = params.get("task_id")
    # Process the agent task
    return {
        "task_id": task_id,
        "status": "completed",
        "result": "Task processed successfully"
    }

# Handle A2A communication
async def handle_agent_request(request_data):
    response = await protocol._handle_raw_async(request_data)
    return response

Future of AI Agent Interoperability

The convergence of JSON-RPC with AI agent protocols like A2A represents a significant step toward true multi-agent ecosystems. As organizations deploy increasingly sophisticated AI systems, these standardized communication protocols enable:

• Cross-platform agent collaboration regardless of underlying frameworks
• Scalable enterprise AI workflows with secure inter-agent communication
• Modular AI architectures where specialized agents can be dynamically combined
• Vendor-neutral AI ecosystems reducing lock-in and increasing flexibility

The adoption of JSON-RPC as the foundation for A2A and similar protocols demonstrates how established web standards can be effectively adapted to meet the unique requirements of AI agent communication, providing a solid technical foundation for the next generation of collaborative AI systems.

Practical Implementation Resources

For comprehensive Python-based examples and implementations of JSON-RPC, A2A Protocol, and MCP communication patterns, including working code samples, test suites, and detailed documentation, visit the AI Agents Basics repository. This resource provides production-ready implementations that demonstrate best practices for building interoperable AI agent systems.

This section demonstrates a complete A2A (Agent-to-Agent) protocol implementation featuring:

• A tiny A2A server in Python that wraps a CrewAI mini-crew
• An AutoGen client tool that calls message/send on that server
• The Agent Card published at /.well-known/agent-card.json

A2A Protocol Highlights

• One HTTP endpoint that implements JSON-RPC methods like message/send and message/stream (SSE)
• Messages carry role and parts (e.g., TextPart) and return either a Message or a Task
• Public discovery via an Agent Card that declares URL, transport, skills, and auth at /.well-known/agent-card.json

1) Minimal A2A Server (FastAPI + CrewAI)

Creates a single JSON-RPC endpoint /a2a/jsonrpc that implements message/send (sync) and message/stream (SSE). Internally, a tiny CrewAI "Researcher → Writer" pipeline answers the prompt.

# server.py
import os, uuid, json, asyncio
from typing import AsyncGenerator, Dict, Any
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel
# pip install fastapi uvicorn crewai sse-starlette (or starlette>=0.36)
from crewai import Agent, Task, Crew

# -------- A2A data models (minimal subset) ----------
class TextPart(BaseModel):
    type: str = "text"
    text: str

class Message(BaseModel):
    role: str  # "user" or "agent"
    parts: list[TextPart]
    taskId: str | None = None  # optional, for continuing a task

class MessageSendConfiguration(BaseModel):
    acceptedOutputModes: list[str] | None = None
    historyLength: int | None = None

class MessageSendParams(BaseModel):
    message: Message
    configuration: MessageSendConfiguration | None = None
    metadata: Dict[str, Any] | None = None

class JSONRPCRequest(BaseModel):
    jsonrpc: str
    id: str | int | None
    method: str
    params: Dict[str, Any] | None = None

# -------- CrewAI mini-crew ----------
def run_crewai_pipeline(user_text: str) -> str:
    # Expect OPENAI_API_KEY (or configure your LLM of choice)
    researcher = Agent(
        role="Researcher",
        goal="Find 3 crisp bullet points answering the question.",
        backstory="You scan reliable sources and synthesize insights.",
        allow_code_execution=False,
        verbose=False,
    )
    writer = Agent(
        role="Writer",
        goal="Summarize clearly in <=120 words.",
        backstory="You write concise, structured summaries.",
        allow_code_execution=False,
        verbose=False,
    )
    t1 = Task(description=f"Research the following question and produce 3 bullets:\n{user_text}",
              agent=researcher,
              expected_output="Exactly 3 bullet points.")
    t2 = Task(description="Turn the bullets into a 120-word answer.",
              agent=writer,
              context=[t1],
              expected_output="<=120 words summary.")
    crew = Crew(agents=[researcher, writer], tasks=[t1, t2])
    result = crew.kickoff()  # typically returns the last task's output
    return str(result)

# -------- FastAPI app ----------
app = FastAPI()

@app.post("/a2a/jsonrpc")
async def a2a_jsonrpc(req: Request):
    body = await req.json()
    rpc = JSONRPCRequest(**body)
    method = rpc.method
    params = rpc.params or {}

    # message/send (sync) -> returns a Message or Task (we'll return a Message)
    if method == "message/send":
        p = MessageSendParams(**params)
        # Extract plain text from the first TextPart
        user_text = next((pr.text for pr in p.message.parts if pr.type == "text"), "")
        answer = run_crewai_pipeline(user_text)
        msg = {
            "role": "agent",
            "parts": [{"type":"text","text": answer}],
            # Optionally include a taskId if you manage state
        }
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": rpc.id,
            "result": {"message": msg}
        })

    # message/stream -> SSE stream of SendStreamingMessageResponse events
    if method == "message/stream":
        p = MessageSendParams(**params)
        user_text = next((pr.text for pr in p.message.parts if pr.type == "text"), "")
        task_id = str(uuid.uuid4())

        async def event_stream() -> AsyncGenerator[bytes, None]:
            # 1) Task status: RUNNING
            status_ev = {
                "jsonrpc":"2.0","id":rpc.id,
                "result":{
                    "event":"TaskStatusUpdateEvent",
                    "taskId": task_id,
                    "status":{"state":"running"}  # minimal
                }
            }
            yield f"data: {json.dumps(status_ev)}\n\n".encode()

            # 2) Fake incremental chunks (you can break CrewAI output into chunks if desired)
            await asyncio.sleep(0.2)
            chunk1 = {"jsonrpc":"2.0","id":rpc.id,
                      "result":{"event":"TaskArtifactUpdateEvent","taskId":task_id,
                                "artifact":{"parts":[{"type":"text","text":"Working on it..."}], "append":True}}}
            yield f"data: {json.dumps(chunk1)}\n\n".encode()

            # 3) Final answer
            answer = run_crewai_pipeline(user_text)
            await asyncio.sleep(0.1)
            chunk2 = {"jsonrpc":"2.0","id":rpc.id,
                      "result":{"event":"TaskArtifactUpdateEvent","taskId":task_id,
                                "artifact":{"parts":[{"type":"text","text":answer}], "final":True}}}
            yield f"data: {json.dumps(chunk2)}\n\n".encode()

            # 4) Task status: COMPLETED
            done_ev = {"jsonrpc":"2.0","id":rpc.id,
                       "result":{"event":"TaskStatusUpdateEvent","taskId":task_id,
                                 "status":{"state":"completed"}}}
            yield f"data: {json.dumps(done_ev)}\n\n".encode()

        return StreamingResponse(event_stream(), media_type="text/event-stream")

    # Unknown method -> JSON-RPC error
    return JSONResponse({
        "jsonrpc":"2.0","id": rpc.id,
        "error":{"code": -32601, "message": f"Method not found: {method}"}
    }, status_code=400)

Running the Server

uvicorn server:app --reload --port 8080

Quick Test (Sync)

curl -s http://localhost:8080/a2a/jsonrpc \
  -H "Content-Type: application/json" \
  -d @- <<'JSON'
{"jsonrpc":"2.0","id":1,"method":"message/send",
 "params":{"message":{"role":"user","parts":[{"type":"text","text":"Explain A2A briefly"}]}}}
JSON

The message/send and message/stream naming follow the spec; streaming uses SSE with JSON-RPC responses.

2) Agent Card (Publish for Discovery)

Save as public/.well-known/agent-card.json (or serve at that path). It declares where to call, preferred transport, auth, skills, and modes.

{
  "protocolVersion": "0.3.0",
  "name": "CrewAI Research & Write",
  "description": "Researches a question and returns a concise summary.",
  "url": "http://localhost:8080/a2a/jsonrpc",
  "preferredTransport": "jsonrpc",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "skills": [
    {
      "id": "research_write.v1",
      "name": "Research and summarize",
      "inputModes": ["text/plain"],
      "outputModes": ["text/plain"]
    }
  ],
  "securitySchemes": [
    { "type": "none", "name": "public" }
  ],
  "security": [{ "scheme": "public" }]
}

The spec requires an Agent Card and recommends the well-known path. It also defines fields like protocolVersion, url, preferredTransport, skills, securitySchemes.

3) AutoGen Client: Call Your A2A Agent as a Tool

We register a small FunctionTool that POSTs a JSON-RPC message/send with a TextPart, then the AssistantAgent can call it in-loop. AutoGen includes a tool system and an HTTP tool family; here we show a direct function tool for clarity.

# autogen_client.py
import httpx, asyncio, json
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.tools import FunctionTool

A2A_URL = "http://localhost:8080/a2a/jsonrpc"

async def a2a_send(prompt: str) -> str:
    """Send a prompt to the A2A agent and return text reply."""
    payload = {
        "jsonrpc": "2.0",
        "id": "cli-1",
        "method": "message/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": prompt}]
            }
        }
    }
    async with httpx.AsyncClient(timeout=120) as client:
        r = await client.post(A2A_URL, json=payload)
        r.raise_for_status()
        data = r.json()
        # Per spec, result can be {message} or {task}; we handle {message}.
        return data["result"]["message"]["parts"][0]["text"]

async def main():
    tool = FunctionTool(a2a_send, description="Call remote CrewAI agent via A2A")
    model = OpenAIChatCompletionClient(model="gpt-4o-mini")  # any supported model
    agent = AssistantAgent(
        name="autogen-client",
        model_client=model,
        tools=[tool],
        system_message="Use the tool when you need external research+summary."
    )
    res = await agent.run(task="Summarize the benefits of the A2A protocol.")
    print(res.messages[-1].content)

if __name__ == "__main__":
    asyncio.run(main())

AutoGen's AssistantAgent can use Python FunctionTools; we convert a tool call into an A2A message/send over HTTP. Built-in HTTP/MCP workbenches exist too, but a custom FunctionTool keeps it explicit.

Why This is "A2A-Compliant Enough" for a Starter

• Transport & Methods: We expose JSON-RPC with message/send, and for live tokens we offer message/stream via SSE, matching the spec's streaming rules
• Message shape: The client sends a Message with role and TextPart; server returns a Message (or could return a Task if you adopt long-running polling)
• Discovery: Publishing an Agent Card lets AutoGen (or other clients) discover url, transport choice, skills, and auth scheme

Production Hardening Checklist (Quick)

• Auth: Replace security: public with OAuth2/JWT/Bearer; enforce per the card
• Stateful tasks: Return taskId and implement tasks/get, tasks/cancel, and push notifications if you need webhooks
• Streaming fidelity: Emit TaskStatusUpdateEvent + TaskArtifactUpdateEvent per spec while CrewAI produces chunks
• AgentCard versioning: Keep protocolVersion aligned with the spec you target

Key Benefits of This Implementation

• Standards Compliance: Follows A2A protocol specifications for agent-to-agent communication
• Framework Integration: Seamlessly combines CrewAI's multi-agent capabilities with AutoGen's conversational AI
• Scalable Architecture: Supports both synchronous and asynchronous communication patterns
• Discovery Mechanism: Agent Card enables automatic discovery and integration by other agents
• Streaming Support: Real-time communication via Server-Sent Events for long-running tasks