Friday, May 1, 2026

Building a Python Async Client for an MCP Server

 MCP (Model Context Protocol) is gaining traction as a standard way for AI tools to call external services. Most examples show MCP clients in TypeScript. Here's how I built one in Python, wiring an npm-based MCP server into a FastAPI async backend.

The Idea

The MCP server I needed wraps the Jira API -- it exposes tools like jira_search_issues, jira_create_issue, and jira_get_transitions. The server speaks JSON-RPC 2.0 over stdio: you spawn it as a subprocess, write requests to its stdin, and read responses from its stdout.

Spawning the Server

Python's asyncio.create_subprocess_exec handles this cleanly:

self._proc = await asyncio.create_subprocess_exec(
    "npx", "--yes", "@xuandev/atlassian-mcp",
    stdin=asyncio.subprocess.PIPE,
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
    env=env,
)

Credentials (API token, domain, email) go into the env dict -- no hardcoding, no extra config files.

The MCP Handshake

Before calling any tools, MCP requires an initialization exchange. Send an initialize request, get capabilities back, then send a notifications/initialized one-way notification. Only after this handshake are tool calls available.

Request/Response Correlation

The tricky part with stdio JSON-RPC is that notifications can arrive interleaved with responses. The solution: attach an incrementing id to every request and skip responses until the matching id comes back.

async def _rpc(self, method, params, timeout=30.0):
    msg_id = self._next_id
    self._next_id += 1
    await self._send({"jsonrpc": "2.0", "id": msg_id, "method": method, "params": params})
    while True:
        resp = await self._recv(timeout=timeout)
        if resp.get("id") != msg_id:
            continue
        if "error" in resp:
            raise JiraMcpError(f"MCP error: {resp['error']}")
        return resp.get("result")

Each _recv awaits proc.stdout.readline(), so the loop yields to the event loop between reads -- no busy-waiting.

Lifecycle as a Context Manager

Wrapping everything in an async context manager keeps usage clean:

async with JiraMcpClient() as client:
    issues = await client.search_issues("project = MYPROJ ORDER BY priority ASC")
    new_key = await client.create_issue("MYPROJ", "Fix login bug", "Steps to reproduce...")

The subprocess spawns on __aenter__ and gets killed on __aexit__, exception or not.

What I Learned

The stdio transport is underrated. No ports, no HTTP server, no authentication layer -- just a process. If you need to integrate an MCP-compatible server into a backend that already uses asyncio, this pattern works well and is surprisingly easy to test: mock the subprocess with AsyncMock and feed it canned JSON-RPC responses.

The full client -- handshake, tool dispatch, response parsing, feature-flag gating -- came in under 200 lines of Python. That's a reasonable price for full Jira integration without touching the REST API directly.