Building a Custom MCP Client with the Agent SDK
Every MCP server needs something on the other end of the connection.
That something is a client, and the same MCP Python SDK that ships server-building primitives also ships client primitives: connect to a server, list what it offers, call a tool, read a resource.
When that client lives inside an application built on the Claude Agent SDK, the tools an MCP server exposes become tools Claude itself can reach for during a conversation.
Summary
A custom MCP client connects to a server over a transport, most commonly stdio_client for a local server process.
The connection produces a ClientSession, the object you use to negotiate capabilities and then send requests.
list_tools(), call_tool(), and read_resource() are the three primitives that cover the everyday workflow: discover what's available, invoke it, and read data.
A custom client is useful any time you need programmatic access to an MCP server outside of Claude Desktop or Claude Code, for example wiring an MCP server's tools into an application built on the Agent SDK.
This page walks through the client side of the same request lifecycle covered on the request-handling explainer page.
Recipe
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
server_params = StdioServerParameters(command="python", args=["server.py"])
async def main():
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
tools = await session.list_tools()
print([t.name for t in tools.tools])
asyncio.run(main())When to reach for this:
- You're building an application that needs to call MCP tools directly, outside of Claude Desktop or Claude Code.
- You want to wire an MCP server's tools into a Claude Agent SDK application as part of a larger agent loop.
- You need to write integration tests that exercise a real server through the same code path a production client uses.
- You're debugging a server and want to see exactly what it reports during capability negotiation.
Working Example
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
server_params = StdioServerParameters(
command="python",
args=["tickets_server.py"],
)
async def run_client() -> None:
async with stdio_client(server_params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
tools_result = await session.list_tools()
print("Available tools:")
for tool in tools_result.tools:
print(f" - {tool.name}: {tool.description}")
create_result = await session.call_tool(
"create_ticket",
arguments={"title": "Printer offline", "priority": "low"},
)
ticket_id = create_result.content[0].text
print(f"Created ticket: {ticket_id}")
resource_result = await session.read_resource("tickets://open")
print("Open tickets:")
print(resource_result.contents[0].text)
close_result = await session.call_tool(
"close_ticket",
arguments={"ticket_id": ticket_id},
)
print(close_result.content[0].text)
if __name__ == "__main__":
asyncio.run(run_client())What this demonstrates:
- The full connect-negotiate-exchange lifecycle from the client side:
stdio_clientopens the transport,ClientSessionnegotiates,initialize()completes the handshake. list_tools()used to discover what the server offers before assuming any tool exists.call_tool()invoked twice, once to create a ticket and once to close it, showing how a client drives a multi-step workflow.read_resource()used to fetch read-only data by URI, distinct from the action-oriented tool calls around it.
Deep Dive
How It Works
stdio_client(server_params)spawns the server as a subprocess using the given command and arguments, and returns a pair of streams for reading from and writing to it.ClientSession(read_stream, write_stream)wraps those streams in the protocol layer, giving you the request/response primitives instead of raw bytes.session.initialize()performs capability negotiation, the client and server exchange supported versions, and the server returns its manifest of tools, resources, and prompts.- Every call after
initialize(),list_tools(),call_tool(),read_resource(), is a single request/response cycle over the already-negotiated session. - Both
async withblocks handle cleanup: closing the session and terminating the subprocess when the block exits, even if an exception is raised inside it.
Client Primitives at a Glance
| Primitive | Purpose | Returns |
|---|---|---|
list_tools() | Discover available tools | A list of tool definitions with name, description, input schema |
call_tool(name, arguments) | Invoke a tool | A result with a content list |
read_resource(uri) | Fetch resource content | A result with a contents list |
list_resources() | Discover available resources | A list of resource definitions |
Handling Tool Errors
from mcp import McpError
try:
result = await session.call_tool("close_ticket", arguments={"ticket_id": "T-999"})
except McpError as exc:
print(f"tool call failed: {exc}")A tool call that raises an exception on the server side surfaces to the client as a structured error rather than a crash, catch it the same way you'd catch any expected failure mode in your application.
Python Notes
# Always call session.initialize() before any other session method.
# Calling list_tools() or call_tool() before initialization completes
# will fail, since the client hasn't yet negotiated what the server offers.
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()
# session is now safe to useParameters & Return Values
| Method | Key Argument | Notes |
|---|---|---|
call_tool(name, arguments) | arguments: dict | Must match the tool's declared input schema |
read_resource(uri) | uri: str | Must match a URI the server actually registered |
list_tools() | none | Safe to call repeatedly, reflects the server's current manifest |
Gotchas
- Calling
list_tools()orcall_tool()beforeinitialize(). The session hasn't negotiated capabilities yet, so the server has no context for the request. Fix: alwaysawait session.initialize()as the first call inside the session block. - Passing arguments that don't match a tool's schema. A typo'd key or wrong type fails validation on the server and returns an error the client must handle. Fix: check
list_tools()output for the exact schema before hardcoding acall_tool()payload. - Not handling
McpErroraroundcall_tool(). An unhandled tool error propagates as an unexpected exception in application code that isn't expecting a protocol-level failure. Fix: wrap tool calls that might fail (invalid IDs, missing records) in a try/except forMcpError. - Leaking the subprocess by skipping the
async withcontext manager. Callingstdio_client()without the context manager can leave the server process running after your script exits. Fix: always useasync with stdio_client(...)so cleanup runs automatically. - Assuming a resource always exists.
read_resource()on a URI the server never registered, or one whose underlying data was removed, raises an error just like an invalid tool call. Fix: calllist_resources()first if you're not certain the URI is still valid. - Reusing one
ClientSessionacross unrelated concurrent tasks without coordination. A session isn't automatically safe for arbitrary concurrent use from multiple coroutines issuing overlapping calls. Fix: serialize calls through one task, or open a session per logical unit of work.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
| Claude Desktop / Claude Code as the client | You just need Claude to use the server interactively, no custom application logic | You need programmatic control, custom orchestration, or automated testing |
| A generic HTTP client against an HTTP/SSE server | The server is deployed remotely and you don't need the stdio subprocess model | The server is local-only and stdio is simpler |
The MCP Inspector (npx @modelcontextprotocol/inspector) | You're manually exploring a server during development | You need this behavior embedded in a running application |
FAQs
What's the minimum code needed to connect to an MCP server?
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()That's the full connect-and-negotiate sequence; everything else builds on top of an initialized session.
Why do I need both stdio_client and ClientSession?
stdio_clienthandles the transport, spawning the subprocess and giving you raw read/write streams.ClientSessionhandles the protocol layer on top of those streams, negotiation, requests, responses.- Separating them lets the same
ClientSessionAPI work over other transports, like HTTP/SSE, without changing your protocol-level code.
How do I know what arguments a tool expects?
Call list_tools() and inspect each tool's input schema. It's the same schema the server generated from its handler's type hints, so it's the authoritative source, not the server's source code.
What happens if I call a tool that doesn't exist?
The call fails with an McpError. Catch it explicitly rather than assuming every call_tool() invocation succeeds, especially if the tool name came from user input or configuration.
Can one client connect to multiple servers at once?
Yes, open a separate stdio_client/ClientSession pair per server. Each session tracks its own negotiated capabilities independently, there's no shared state between them.
How does this relate to the Claude Agent SDK?
An application built on the Agent SDK can use these same client primitives to connect to an MCP server and expose its tools as part of the agent's available toolset, letting Claude call them during a conversation the same way it would any other tool.
Do I need to call list_resources() before read_resource()?
Not strictly, if you already know the URI. It's useful when you don't know what resources exist ahead of time, the same way list_tools() is useful before an unfamiliar call_tool().
What does call_tool() return exactly?
A result object with a content list, typically containing text content blocks. Extract the actual value with something like result.content[0].text, matching the shape the server's handler returned.
Is the client connection synchronous or asynchronous?
Asynchronous. Every primitive, initialize(), list_tools(), call_tool(), read_resource(), is an async method and must be awaited inside an asyncio event loop.
What happens to the server process when my client script exits?
If you used async with stdio_client(...), the context manager terminates the subprocess as part of cleanup when the block exits, including when an exception is raised.
Can I test a server with this client instead of a real deployment?
Yes, that's a common pattern. Connecting a test client to your server over stdio and calling its tools directly is the basis for integration testing, covered in more depth on the dedicated testing page.
Related
- How MCP Servers Handle Requests - the lifecycle this client drives from the other side
- MCP Server Basics - the minimal server this client can connect to
- Building an MCP Server with the Python SDK - the server-side handlers this client calls
- Testing MCP Servers Before Deployment - using a client session inside automated tests
- stdio vs HTTP/SSE MCP Deployment Comparison - connecting to a server over a different transport
Stack versions: Written against the Claude model lineup current as of ~June 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (the default), and Claude Haiku 4.5 - and the current Model Context Protocol Python/TypeScript SDKs. Model names, SDK versions, and the MCP spec move quickly - verify current specifics at platform.claude.com/docs and modelcontextprotocol.io before relying on them.