Building a Custom MCP Client for Your Agent Runtime

While high-level SDK adapters exist for LangChain and LlamaIndex, enterprise platforms building proprietary agent engines often require custom, lightweight MCP clients. Building a custom MCP client gives you complete control over connection management, transport protocol selection (Stdio vs. Server-Sent Events / SSE), error recovery, and security token injection.

In this guide, we walk through building a custom Python MCP client that initializes sessions, lists tools, and executes calls over JSON-RPC.

Understanding the MCP Client Handshake

The Model Context Protocol uses JSON-RPC 2.0 messages over standard I/O (stdio) or HTTP/SSE streams. A client lifecycle follows three distinct phases:

1. Initialization (initialize): Client sends protocol version and capability flags; server responds with capabilities and metadata.
2. Initialized Notification (notifications/initialized): Client acknowledges protocol agreement.
3. Tool Operations (tools/list, tools/call): Client discovers available tool schemas and invokes execution endpoints.

Client                                      MCP Server
  |                                              |
  | -------- initialize (JSON-RPC) ----------->  |
  | <------- initialize result (capabilities) -  |
  | -------- notifications/initialized ------->  |
  |                                              |
  | -------- tools/list ---------------------->  |
  | <------- tools list (schemas) -------------  |
  |                                              |
  | -------- tools/call (args) --------------->  |
  | <------- tools call result ----------------  |

Implementing a Custom Async MCP Client

Using the official low-level mcp library, we construct a reusable MCP client manager:

import asyncio
from typing import Any, Dict, List
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class CustomMCPClient:
    def __init__(self, server_command: str, server_args: List[str]):
        self.server_params = StdioServerParameters(
            command=server_command,
            args=server_args
        )
        self.session: ClientSession = None
        self._exit_stack = None

    async def connect(self):
        """Establish transport and initialize MCP session"""
        from contextlib import AsyncExitStack
        self._exit_stack = AsyncExitStack()
        
        # Connect transport via Stdio
        read, write = await self._exit_stack.enter_async_context(
            stdio_client(self.server_params)
        )
        
        # Initialize Client Session
        self.session = await self._exit_stack.enter_async_context(
            ClientSession(read, write)
        )
        
        # Protocol handshake
        init_result = await self.session.initialize()
        print(f"Connected to MCP Server: {init_result.serverInfo.name} v{init_result.serverInfo.version}")

    async def get_available_tools(self) -> List[Dict[str, Any]]:
        """Fetch advertised tools and format into LLM function-calling schema"""
        response = await self.session.list_tools()
        formatted_tools = []
        for tool in response.tools:
            formatted_tools.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema
                }
            })
        return formatted_tools

    async def execute_tool(self, name: str, arguments: Dict[str, Any]) -> str:
        """Execute a remote MCP tool and extract text output"""
        result = await self.session.call_tool(name, arguments)
        
        output_chunks = []
        for content in result.content:
            if content.type == "text":
                output_chunks.append(content.text)
        return "\n".join(output_chunks)

    async def close(self):
        """Cleanly close transport connections"""
        if self._exit_stack:
            await self._exit_stack.aclose()

Integrating with Custom Agent Loops

async def main():
    client = CustomMCPClient("python", ["server.py"])
    await client.connect()

    try:
        # Discover tools for LLM prompt
        tools = await client.get_available_tools()
        print("Discovered Tools:", [t["function"]["name"] for t in tools])

        # Execute tool call when directed by LLM
        output = await client.execute_tool("fetch_data", {"query": "AI Agents 2026"})
        print("Tool Output:", output)

    finally:
        await client.close()

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

Key Client Implementation Considerations

• Transport Switching: Ensure your client abstraction supports switching between stdio_client (local subprocesses) and sse_client (remote HTTP servers).
• Graceful Reconnection: Implement exponential backoff reconnection strategies for persistent SSE sessions.
• Request Timeouts: Wrap session.call_tool() invocations in asyncio.wait_for(..., timeout=30.0) to safeguard agent runtimes against stalled tools.

For detailed wire protocol specifications and client implementation guides, consult the official Model Context Protocol Documentation.