Understanding JSON-RPC — The Wire Protocol Behind MCP, ACP and Multi-Agent Systems

Introduction

When I started building MCP (Model Context Protocol) servers and ACP (Agent Communication Protocol) agents, I kept running into the same underlying mechanism: JSON-RPC 2.0. At first it looked like just a thin wrapper around JSON — but once I understood the spec properly I realised why it became the standard wire protocol for agent-to-tool and agent-to-agent communication. It is stateless, transport-agnostic, trivially debuggable, and adds just enough structure to distinguish a call from a fire-and-forget notification.

This article covers the full JSON-RPC 2.0 specification, implements a working server and client in Python 3.12, then looks at how MCP, ACP, and multi-agent frameworks rely on it in practice.

---

What is JSON-RPC?

JSON-RPC is a stateless, lightweight Remote Procedure Call (RPC) protocol defined by the JSON-RPC Working Group. The current version is 2.0 (2010, revised 2013).

Its design constraints are intentionally minimal:

• Transport-agnostic — works over stdio, HTTP, WebSockets, Unix sockets, or in-process

• Format — plain JSON (RFC 4627), no binary encoding

• Direction — a client sends a request, a server returns a response; both roles can be held by the same process simultaneously

• Versions — "jsonrpc": "2.0" is always present; 1.0 had no version field

That last point matters a lot for agent systems: an LLM host can be both a JSON-RPC server (answering tool calls from an orchestrator) and a JSON-RPC client (calling out to MCP servers) at the same time, over the same connection.

---

The Specification

Request Object

A request has four possible members:

Field	Required	Description
jsonrpc	Yes	Always the string "2.0"
method	Yes	String name of the method to invoke
params	No	Array (positional) or Object (named)
id	No	String, Number, or Null — omitting it makes it a Notification

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "dns_lookup",
    "arguments": { "hostname": "example.com" }
  },
  "id": 1
}

Response Object

The server must reply to every request (non-Notification). Exactly one of result or error is set — never both.

{
  "jsonrpc": "2.0",
  "result": {
    "content": [{ "type": "text", "text": "93.184.216.34" }]
  },
  "id": 1
}

Notification

A request without id. The server must not send a response. Fire-and-forget:

{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": { "progressToken": "abc", "progress": 42, "total": 100 }
}

MCP uses notifications heavily — for progress updates, resource change events, and log messages — so the client never blocks waiting for an acknowledgement.

Error Object

When something goes wrong the response carries an error instead of a result:

{
  "jsonrpc": "2.0",
  "error": {
    "code": -32601,
    "message": "Method not found",
    "data": { "method": "tools/nonexistent" }
  },
  "id": 1
}

Reserved error codes:

Code	Name	Meaning
-32700	Parse error	Invalid JSON received
-32600	Invalid Request	Not a valid Request object
-32601	Method not found	Method does not exist
-32602	Invalid params	Invalid method parameters
-32603	Internal error	Internal JSON-RPC error
-32000 to -32099	Server error	Application-defined server errors

Batch Requests

A client can send an Array of Request objects in one call. The server responds with a matching Array (notifications get no slot in the response array):

// Request
[
  {"jsonrpc": "2.0", "method": "resources/list", "id": 1},
  {"jsonrpc": "2.0", "method": "tools/list",     "id": 2},
  {"jsonrpc": "2.0", "method": "notifications/initialized"}
]

// Response (notification produces no entry)
[
  {"jsonrpc": "2.0", "result": {"resources": []}, "id": 1},
  {"jsonrpc": "2.0", "result": {"tools": [...]},  "id": 2}
]

---

Python 3.12 Implementation

Data Models

# jsonrpc/models.py
from __future__ import annotations
from typing import Any
from pydantic import BaseModel, Field, model_validator

Id = str | int | None

class JsonRpcRequest(BaseModel):
    jsonrpc: str = "2.0"
    method: str
    params: dict | list | None = None
    id: Id = None

    @property
    def is_notification(self) -> bool:
        return self.id is None

    @model_validator(mode="after")
    def validate_version(self) -> "JsonRpcRequest":
        if self.jsonrpc != "2.0":
            raise ValueError("Only JSON-RPC 2.0 is supported")
        return self


class JsonRpcError(BaseModel):
    code: int
    message: str
    data: Any = None


class JsonRpcResponse(BaseModel):
    jsonrpc: str = "2.0"
    id: Id
    result: Any = None
    error: JsonRpcError | None = None

    @model_validator(mode="after")
    def only_result_or_error(self) -> "JsonRpcResponse":
        if self.result is not None and self.error is not None:
            raise ValueError("result and error are mutually exclusive")
        return self


# Standard error constructors
def parse_error(id: Id = None) -> JsonRpcResponse:
    return JsonRpcResponse(id=id, error=JsonRpcError(code=-32700, message="Parse error"))

def invalid_request(id: Id = None) -> JsonRpcResponse:
    return JsonRpcResponse(id=id, error=JsonRpcError(code=-32600, message="Invalid Request"))

def method_not_found(id: Id, method: str) -> JsonRpcResponse:
    return JsonRpcResponse(
        id=id,
        error=JsonRpcError(code=-32601, message="Method not found",
                           data={"method": method}),
    )

def invalid_params(id: Id, detail: str) -> JsonRpcResponse:
    return JsonRpcResponse(
        id=id,
        error=JsonRpcError(code=-32602, message="Invalid params", data=detail),
    )

def internal_error(id: Id, detail: str | None = None) -> JsonRpcResponse:
    return JsonRpcResponse(
        id=id,
        error=JsonRpcError(code=-32603, message="Internal error", data=detail),
    )

Minimal JSON-RPC Server

# jsonrpc/server.py
from __future__ import annotations
import asyncio
import json
import sys
from collections.abc import Callable, Awaitable
from typing import Any
from .models import (
    JsonRpcRequest, JsonRpcResponse,
    method_not_found, parse_error, invalid_request, internal_error,
)

Handler = Callable[[dict | list | None], Awaitable[Any]]

class JsonRpcServer:
    """
    Minimal JSON-RPC 2.0 server over stdio.
    Matches the transport used by MCP and ACP servers.
    """

    def __init__(self) -> None:
        self._methods: dict[str, Handler] = {}

    def method(self, name: str) -> Callable[[Handler], Handler]:
        """Decorator to register a method handler."""
        def decorator(fn: Handler) -> Handler:
            self._methods[name] = fn
            return fn
        return decorator

    async def _handle_single(self, raw: dict) -> JsonRpcResponse | None:
        try:
            request = JsonRpcRequest.model_validate(raw)
        except Exception:
            return invalid_request(raw.get("id"))

        handler = self._methods.get(request.method)
        if handler is None:
            if request.is_notification:
                return None     # notifications produce no response regardless
            return method_not_found(request.id, request.method)

        try:
            result = await handler(request.params)
        except Exception as exc:
            if request.is_notification:
                return None
            return internal_error(request.id, str(exc))

        if request.is_notification:
            return None

        return JsonRpcResponse(id=request.id, result=result)

    async def _handle_message(self, line: str) -> list[JsonRpcResponse]:
        try:
            data = json.loads(line)
        except json.JSONDecodeError:
            return [parse_error()]

        if isinstance(data, list):
            if not data:
                return [invalid_request()]
            responses = await asyncio.gather(*[
                self._handle_single(item) if isinstance(item, dict)
                else asyncio.coroutine(lambda: invalid_request())()
                for item in data
            ])
            return [r for r in responses if r is not None]

        if not isinstance(data, dict):
            return [invalid_request()]

        response = await self._handle_single(data)
        return [response] if response is not None else []

    async def run_stdio(self) -> None:
        """Read newline-delimited JSON from stdin, write responses to stdout."""
        reader = asyncio.StreamReader()
        protocol = asyncio.StreamReaderProtocol(reader)
        loop = asyncio.get_running_loop()
        await loop.connect_read_pipe(lambda: protocol, sys.stdin)
        writer_transport, writer_protocol = await loop.connect_write_pipe(
            asyncio.BaseProtocol, sys.stdout.buffer
        )
        writer = asyncio.StreamWriter(writer_transport, writer_protocol, reader, loop)

        while True:
            try:
                line = await reader.readline()
            except Exception:
                break
            if not line:
                break

            responses = await self._handle_message(line.decode().strip())
            for resp in responses:
                payload = resp.model_dump_json(exclude_none=False) + "\n"
                writer.write(payload.encode())
                await writer.drain()

Registering Methods

# server_example.py
import asyncio
import socket
from jsonrpc.server import JsonRpcServer

server = JsonRpcServer()

@server.method("ping")
async def ping(params: dict | list | None) -> dict:
    return {"pong": True}

@server.method("dns/lookup")
async def dns_lookup(params: dict | list | None) -> dict:
    if not isinstance(params, dict) or "hostname" not in params:
        raise ValueError("params.hostname is required")
    hostname = params["hostname"]
    try:
        loop = asyncio.get_running_loop()
        infos = await loop.getaddrinfo(hostname, None)
        addresses = list({info[4][0] for info in infos})
        return {"hostname": hostname, "addresses": addresses}
    except socket.gaierror as exc:
        raise ValueError(f"DNS resolution failed: {exc}") from exc

@server.method("compliance/run")
async def compliance_run(params: dict | list | None) -> dict:
    """
    Calls ansible-inspec to run a compliance profile against a set of hosts.
    Real implementation shells out to ansible-playbook.
    """
    if not isinstance(params, dict):
        raise ValueError("params must be an object")
    profile = params.get("profile", "dev-sec/linux-baseline")
    hosts = params.get("hosts", [])
    # In the real ansible-inspec server this dispatches a background job
    return {"job_id": "abc-123", "profile": profile, "hosts": hosts, "status": "pending"}

asyncio.run(server.run_stdio())

Client

# jsonrpc/client.py
from __future__ import annotations
import asyncio
import json
import uuid
from .models import JsonRpcRequest, JsonRpcResponse

class JsonRpcClient:
    """Async JSON-RPC 2.0 client over stdio (subprocess)."""

    def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        self._reader = reader
        self._writer = writer
        self._pending: dict[str | int, asyncio.Future] = {}
        self._recv_task: asyncio.Task | None = None

    async def start(self) -> None:
        self._recv_task = asyncio.create_task(self._receive_loop())

    async def _receive_loop(self) -> None:
        while True:
            line = await self._reader.readline()
            if not line:
                break
            try:
                data = json.loads(line)
                response = JsonRpcResponse.model_validate(data)
                future = self._pending.pop(response.id, None)
                if future and not future.done():
                    future.set_result(response)
            except Exception:
                pass

    async def call(self, method: str, params: dict | list | None = None) -> JsonRpcResponse:
        """Send a request and wait for the response."""
        request_id = str(uuid.uuid4())[:8]
        request = JsonRpcRequest(method=method, params=params, id=request_id)
        future: asyncio.Future[JsonRpcResponse] = asyncio.get_event_loop().create_future()
        self._pending[request_id] = future

        self._writer.write((request.model_dump_json() + "\n").encode())
        await self._writer.drain()

        return await asyncio.wait_for(future, timeout=30.0)

    async def notify(self, method: str, params: dict | list | None = None) -> None:
        """Send a notification — no response expected."""
        notification = JsonRpcRequest(method=method, params=params)  # no id
        self._writer.write((notification.model_dump_json() + "\n").encode())
        await self._writer.drain()


async def launch_server(command: list[str]) -> JsonRpcClient:
    """Spawn a server process and wrap it in a JSON-RPC client."""
    proc = await asyncio.create_subprocess_exec(
        *command,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
    )
    client = JsonRpcClient(proc.stdout, proc.stdin)
    await client.start()
    return client


# Usage
async def main() -> None:
    client = await launch_server(["python", "server_example.py"])

    # Regular call
    response = await client.call("ping")
    print(response.result)   # {"pong": True}

    # Call with params
    response = await client.call("dns/lookup", {"hostname": "github.com"})
    if response.error:
        print(f"Error: {response.error.message}")
    else:
        print(response.result)   # {"hostname": "github.com", "addresses": [...]}

    # Notification — no response
    await client.notify("notifications/log", {"level": "info", "message": "client started"})

asyncio.run(main())

---

JSON-RPC in Modern Agent Systems

MCP — Model Context Protocol

MCP is Anthropic's open protocol for connecting LLMs to tools, resources, and prompts. Its wire format is JSON-RPC 2.0 over stdio (for local servers) or HTTP + SSE (for remote servers).

Every interaction is a JSON-RPC call:

// Host → MCP server: list available tools
{"jsonrpc": "2.0", "method": "tools/list", "id": 1}

// MCP server → Host: response
{
  "jsonrpc": "2.0",
  "result": {
    "tools": [{
      "name": "dns_lookup",
      "description": "Resolve a hostname to IP addresses",
      "inputSchema": {
        "type": "object",
        "properties": { "hostname": { "type": "string" } },
        "required": ["hostname"]
      }
    }]
  },
  "id": 1
}

// Host → MCP server: invoke the tool
{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": { "name": "dns_lookup", "arguments": { "hostname": "github.com" } },
  "id": 2
}

// MCP server → Host: tool result
{
  "jsonrpc": "2.0",
  "result": {
    "content": [{"type": "text", "text": "140.82.112.4, 140.82.113.4"}],
    "isError": false
  },
  "id": 2
}

MCP also uses notifications (no id) for:

• notifications/progress — long-running tool keeps the host informed

• notifications/resources/list_changed — resource list changed, host should re-fetch

• notifications/message — server-side log messages

The initialisation handshake is itself a JSON-RPC exchange:

// Host → Server
{
  "jsonrpc": "2.0",
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": { "roots": { "listChanged": true } },
    "clientInfo": { "name": "claude-desktop", "version": "0.7" }
  },
  "id": 0
}

// Server → Host
{
  "jsonrpc": "2.0",
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": { "tools": {} },
    "serverInfo": { "name": "dns-server", "version": "0.1.0" }
  },
  "id": 0
}

// Host → Server (notification, no response expected)
{"jsonrpc": "2.0", "method": "notifications/initialized"}

Python MCP Server with JSON-RPC

Here is a minimal Python MCP server that uses the same JSON-RPC primitives manually (without an SDK), showing the protocol directly:

# mcp_server.py — minimal stdio MCP server in pure Python 3.12
import asyncio
import json
import sys
from typing import Any

TOOLS = [
    {
        "name": "compliance_run",
        "description": "Run an ansible-inspec compliance profile against hosts",
        "inputSchema": {
            "type": "object",
            "properties": {
                "profile": {
                    "type": "string",
                    "description": "Profile name, e.g. dev-sec/linux-baseline"
                },
                "hosts": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "List of hostnames to check"
                }
            },
            "required": ["profile", "hosts"]
        }
    }
]

def rpc_response(id: Any, result: Any) -> str:
    return json.dumps({"jsonrpc": "2.0", "result": result, "id": id})

def rpc_error(id: Any, code: int, message: str) -> str:
    return json.dumps({"jsonrpc": "2.0", "error": {"code": code, "message": message}, "id": id})

def rpc_notification(method: str, params: dict) -> str:
    return json.dumps({"jsonrpc": "2.0", "method": method, "params": params})

async def handle(msg: dict) -> str | None:
    method = msg.get("method", "")
    id_ = msg.get("id")
    params = msg.get("params", {})
    is_notification = id_ is None

    match method:
        case "initialize":
            return rpc_response(id_, {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "ansible-inspec-mcp", "version": "0.2.12"}
            })

        case "notifications/initialized":
            return None  # notification — no reply

        case "tools/list":
            return rpc_response(id_, {"tools": TOOLS})

        case "tools/call":
            tool_name = params.get("name")
            arguments = params.get("arguments", {})

            if tool_name == "compliance_run":
                profile = arguments.get("profile", "dev-sec/linux-baseline")
                hosts = arguments.get("hosts", [])
                # Real: dispatch to ansible-inspec job executor
                result_text = f"Queued compliance job: {profile} on {len(hosts)} host(s)"
                return rpc_response(id_, {
                    "content": [{"type": "text", "text": result_text}],
                    "isError": False
                })

            return rpc_error(id_, -32601, f"Unknown tool: {tool_name}")

        case _:
            if is_notification:
                return None
            return rpc_error(id_, -32601, f"Method not found: {method}")


async def main() -> None:
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    loop = asyncio.get_running_loop()
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)

    while True:
        line = await reader.readline()
        if not line:
            break

        try:
            msg = json.loads(line.decode().strip())
        except json.JSONDecodeError:
            out = json.dumps({"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None})
            sys.stdout.write(out + "\n")
            sys.stdout.flush()
            continue

        response = await handle(msg)
        if response is not None:
            sys.stdout.write(response + "\n")
            sys.stdout.flush()

asyncio.run(main())

ACP — Agent Communication Protocol

ACP (from IBM Research) uses JSON-RPC 2.0 over HTTP for agent-to-agent calls. Where MCP connects an LLM host to tools, ACP connects agents to other agents. The framing is the same but the semantics differ:

// Orchestrator → Worker agent: start a run
{
  "jsonrpc": "2.0",
  "method": "runs/create",
  "params": {
    "agent_id": "compliance-checker",
    "input": [
      {
        "parts": [
          {"type": "text", "content": "Run CIS Level 1 benchmark on web-01"}
        ]
      }
    ]
  },
  "id": "run-001"
}

// Worker agent → Orchestrator
{
  "jsonrpc": "2.0",
  "result": {
    "run_id": "run-001",
    "status": "in_progress",
    "agent_id": "compliance-checker"
  },
  "id": "run-001"
}

Progress in ACP comes as a stream: the HTTP response carries Content-Type: text/event-stream and each SSE event is a JSON-RPC notification:

event: message
data: {"jsonrpc":"2.0","method":"runs/status","params":{"run_id":"run-001","status":"in_progress","output":[]}}

event: message
data: {"jsonrpc":"2.0","method":"runs/status","params":{"run_id":"run-001","status":"completed","output":[{"parts":[{"type":"text","content":"3 controls failed"}]}]}}

The caller never blocks — it receives events as they arrive, exactly like the notification stream in stdio MCP.

Multi-Agent Communication

In a multi-agent system (e.g. LangGraph, CrewAI, or a custom orchestrator) JSON-RPC becomes the inter-agent wire format. A typical topology:

Orchestrator agent
    │  JSON-RPC call: "agent/run"
    ├──▶ Compliance agent  (ansible-inspec MCP server)
    │         │  JSON-RPC call: "tools/call"
    │         └──▶ ansible-inspec tool
    │
    └──▶ Report agent
              │  JSON-RPC call: "tools/call"
              └──▶ formatting tool

Python implementation of a simple orchestrator that coordinates agents via JSON-RPC:

# orchestrator.py
import asyncio
from jsonrpc.client import launch_server

async def run_compliance_workflow(
    hosts: list[str],
    profile: str = "dev-sec/linux-baseline",
) -> dict:
    # Launch two MCP servers as sub-processes
    compliance_agent = await launch_server(["python", "compliance_mcp_server.py"])
    report_agent = await launch_server(["python", "report_mcp_server.py"])

    # Initialize both (MCP handshake)
    for agent in [compliance_agent, report_agent]:
        await agent.call("initialize", {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "orchestrator", "version": "1.0"}
        })
        await agent.notify("notifications/initialized")

    # Step 1: run compliance check via compliance agent
    result = await compliance_agent.call("tools/call", {
        "name": "compliance_run",
        "arguments": {"profile": profile, "hosts": hosts}
    })

    if result.error:
        return {"status": "failed", "error": result.error.message}

    raw_output = result.result["content"][0]["text"]

    # Step 2: pass raw output to report agent for formatting
    report = await report_agent.call("tools/call", {
        "name": "format_report",
        "arguments": {
            "raw": raw_output,
            "format": "markdown",
        }
    })

    return {
        "status": "completed",
        "profile": profile,
        "hosts": hosts,
        "report": report.result["content"][0]["text"] if not report.error else None,
    }


asyncio.run(run_compliance_workflow(["web-01", "web-02"], "dev-sec/ssh-baseline"))

Batch Calls for Efficiency

When an orchestrator needs to query capabilities from multiple agents in one shot, batch requests cut round-trips:

import json
import asyncio

async def discover_capabilities(client) -> dict:
    """Query tools, resources, and prompts in one batch call."""
    batch = [
        {"jsonrpc": "2.0", "method": "tools/list",     "id": 1},
        {"jsonrpc": "2.0", "method": "resources/list", "id": 2},
        {"jsonrpc": "2.0", "method": "prompts/list",   "id": 3},
    ]
    # Low-level batch send (most SDK clients expose this)
    client._writer.write((json.dumps(batch) + "\n").encode())
    await client._writer.drain()
    # Collect responses by id...

---

Key Design Observations

Why Stdio for Local Servers?

MCP servers launched by Claude Desktop or VS Code Copilot run as child processes. Stdio is:

• Zero configuration — no port allocation, no firewall rules

• Scoped to the session — process dies with the host

• Synchronous framing — newline-delimited JSON; no HTTP overhead

For remote servers (deployed APIs), MCP uses HTTP POST with a JSON-RPC body plus SSE for streaming notifications. The JSON-RPC payload is identical — only the transport changes, proving the protocol's transport-agnostic design.

Correlation via id

The id field is the only mechanism that correlates a response to its request. In an async server handling 50 concurrent tool calls, each response must carry back the correct id. Missing it means the client cannot match the response to the caller — this is the most common JSON-RPC implementation bug.

Notifications as Event Bus

Because notifications require no response, they behave like a pub/sub channel embedded in the same connection. MCP uses them for:

• notifications/progress — real-time progress of long-running tools

• notifications/resources/updated — push-invalidate cache

• notifications/message — server log forwarding to the host UI

This keeps the single stdio channel busy without ever blocking a pending request.

---

Summary

Concept	Key point
jsonrpc: "2.0"	Always present; distinguishes from 1.0
Request with id	Expects a response
Notification (no id)	No response — fire and forget
result xor error	Exactly one of them in every response
Error codes -32700 to -32603	Reserved; don't reuse for app errors
Batch	Array of requests → Array of responses (notifications excluded)
MCP	JSON-RPC 2.0 over stdio or HTTP+SSE; tools/call, tools/list, notifications
ACP	JSON-RPC 2.0 over HTTP+SSE; runs/create, agent-to-agent
Multi-agent	Orchestrator sends JSON-RPC to each agent as a child process or remote server

JSON-RPC 2.0 is small enough to read in one sitting and solid enough that two of the most significant AI agent protocols of 2025 (MCP and ACP) independently chose it as their wire format. Understanding the spec at the message level — not just through an SDK — makes debugging agent systems significantly easier when things go wrong on the wire.