Building a Custom MCP Client with the Agent SDK
Every MCP server needs something on the other end of the connection.
Search across all documentation pages
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.
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.
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:
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:
stdio_client opens the transport, ClientSession negotiates, 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.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.initialize(), list_tools(), call_tool(), read_resource(), is a single request/response cycle over the already-negotiated session.async with blocks handle cleanup: closing the session and terminating the subprocess when the block exits, even if an exception is raised inside it.| 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 |
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.
# 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 use| 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 |
list_tools() or call_tool() before initialize(). The session hasn't negotiated capabilities yet, so the server has no context for the request. Fix: always await session.initialize() as the first call inside the session block.list_tools() output for the exact schema before hardcoding a call_tool() payload.McpError around call_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 for McpError.async with context manager. Calling stdio_client() without the context manager can leave the server process running after your script exits. Fix: always use async with stdio_client(...) so cleanup runs automatically.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: call list_resources() first if you're not certain the URI is still valid.ClientSession across 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.| 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 |
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.
stdio_client handles the transport, spawning the subprocess and giving you raw read/write streams.ClientSession handles the protocol layer on top of those streams, negotiation, requests, responses.ClientSession API work over other transports, like HTTP/SSE, without changing your protocol-level code.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.
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.
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.
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.
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().
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.
Asynchronous. Every primitive, initialize(), list_tools(), call_tool(), read_resource(), is an async method and must be awaited inside an asyncio event loop.
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.
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.
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.
Reviewed by Chris St. John·Last updated Jul 18, 2026