Build an MCP Client: Python ClientSession Tutorial for AI Apps
TL;DR — Build MCP client in Python with ClientSession and stdio transport. MCP Docs: "Use ClientSession, StdioServerParameters, stdio_client. Connect to server, list tools, call tools, feed results to AI." RealPython: "Build MCP client app for command line. Connect via stdio, list capabilities, use tools for AI-powered chat." MCP Python SDK: "Pin mcp>=1.27,<2. v2 is major rework for 2026-07-28 spec." MCP Info: "Client discovers capabilities through list calls: tools/list, resources/list, prompts/list." LushBinary: "MCP supports two transports. HTTP+SSE deprecated in March 2025 spec. Add AI client, connect to servers, execute tools." Learn more with MCP guide, build MCP server, Claude Desktop MCP, and MCP transports.
MCP Docs provides the foundation: "Build an MCP client that connects to a server, lists available tools, and uses them to answer queries. The client uses the MCP Python SDK with ClientSession, StdioServerParameters, and stdio_client to connect over stdio transport."
RealPython explains the use case: "You will build an MCP client app for the command line using the MCP Python SDK. It will be able to connect to an MCP server through the standard input/output (stdio) transport, list the server's capabilities, and use the server's tools to feed an AI-powered chat."
MCP Client Architecture
'What is the weather in CA?'"] AI["Claude API
Anthropic SDK
receives tools + message"] Decision{"Tool call
needed?"} Execute["Execute Tool
session.call_tool()
via MCP protocol"] Result["Tool Result
feed back to Claude"] Response["Final Response
to user"] end subgraph Session["ClientSession"] Init["session.initialize()
capability negotiation
exchange capabilities"] ListTools["session.list_tools()
discover available tools
name, description, schema"] CallTool["session.call_tool()
invoke tool with args
returns content"] ListRes["session.list_resources()
discover data sources"] ReadRes["session.read_resource()
read resource by URI"] ListPrompts["session.list_prompts()
discover templates"] GetPrompt["session.get_prompt()
get templated message"] end subgraph Transport["stdio Transport"] Params["StdioServerParameters
command, args, env"] StdioClient["stdio_client()
spawns server subprocess
stdin/stdout communication"] ExitStack["AsyncExitStack
manage lifecycle
cleanup on exit"] end subgraph Server["MCP Server (subprocess)"] ServerProc["Server Process
runs via stdio
FastMCP server"] end User --> AI AI --> Decision Decision -->|Yes| Execute Decision -->|No| Response Execute --> CallTool CallTool --> Result Result --> AI AI --> Response Init --> ListTools ListTools --> AI CallTool --> Execute Params --> StdioClient StdioClient --> ExitStack StdioClient -->|stdin/stdout| ServerProc ExitStack --> Session
Client Operation Comparison
| Operation | Method | Returns | When to Use |
|---|---|---|---|
| Initialize | session.initialize() |
Server capabilities | First call after connect |
| List tools | session.list_tools() |
Tool schemas | Discover available tools |
| Call tool | session.call_tool(name, args) |
Content list | Execute AI-requested tool |
| List resources | session.list_resources() |
Resource URIs | Discover data sources |
| Read resource | session.read_resource(uri) |
Resource content | Access read-only data |
| List prompts | session.list_prompts() |
Prompt schemas | Discover templates |
| Get prompt | session.get_prompt(name, args) |
Templated message | Use reusable templates |
Implementation
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class ClientOperation(Enum):
INITIALIZE = "initialize"
LIST_TOOLS = "list_tools"
CALL_TOOL = "call_tool"
LIST_RESOURCES = "list_resources"
READ_RESOURCE = "read_resource"
LIST_PROMPTS = "list_prompts"
GET_PROMPT = "get_prompt"
@dataclass
class MCPClientBuilder:
"""Complete MCP client implementation guide."""
def get_install(self) -> str:
"""Installation instructions."""
return (
"# Install MCP Python SDK and Anthropic SDK\n"
"pip install 'mcp>=1.27,<2' anthropic python-dotenv\n"
"\n"
"# Pin below 2.x while v2 is in alpha\n"
"# v2 is major rework for 2026-07-28 spec"
)
def get_basic_client(self) -> str:
"""Basic MCP client connecting via stdio."""
return (
"import asyncio\n"
"from contextlib import AsyncExitStack\n"
"from mcp import ClientSession, StdioServerParameters\n"
"from mcp.client.stdio import stdio_client\n"
"\n"
"async def main():\n"
" # Server parameters for stdio transport\n"
" server_params = StdioServerParameters(\n"
" command='python',\n"
" args=['server.py'],\n"
" env=None, # inherit environment\n"
" )\n"
" \n"
" # Use AsyncExitStack for resource management\n"
" async with AsyncExitStack() as exit_stack:\n"
" # Connect to server via stdio\n"
" read, write = await exit_stack\\\n"
" .enter_async_context(\n"
" stdio_client(server_params))\n"
" \n"
" # Create client session\n"
" session = await exit_stack\\\n"
" .enter_async_context(\n"
" ClientSession(read, write))\n"
" \n"
" # Initialize — capability negotiation\n"
" await session.initialize()\n"
" \n"
" # List available tools\n"
" tools = await session.list_tools()\n"
" print('Available tools:')\n"
" for tool in tools.tools:\n"
" print(f' - {tool.name}: '\n"
" f'{tool.description}')\n"
" \n"
" # Call a tool\n"
" result = await session.call_tool(\n"
" 'get_data',\n"
" {'query': 'test query'}\n"
" )\n"
" print(f'Result: {result.content}')\n"
"\n"
"if __name__ == '__main__':\n"
" asyncio.run(main())"
)
def get_full_client(self) -> str:
"""Full MCP client with Claude API integration."""
return (
"import asyncio\n"
"import sys\n"
"from contextlib import AsyncExitStack\n"
"from typing import Optional\n"
"\n"
"from mcp import ClientSession,\n"
" StdioServerParameters\n"
"from mcp.client.stdio import stdio_client\n"
"\n"
"from anthropic import Anthropic\n"
"from dotenv import load_dotenv\n"
"\n"
"load_dotenv()\n"
"\n"
"class MCPClient:\n"
" def __init__(self):\n"
" self.session: Optional[\n"
" ClientSession] = None\n"
" self.exit_stack = AsyncExitStack()\n"
" self.anthropic = Anthropic()\n"
" \n"
" async def connect_to_server(\n"
" self, server_script: str\n"
" ):\n"
" '''Connect to MCP server.'''\n"
" server_params = StdioServerParameters(\n"
" command=sys.executable,\n"
" args=[server_script],\n"
" env=None,\n"
" )\n"
" \n"
" # stdio transport\n"
" stdio_transport = await \\\n"
" self.exit_stack\\\n"
" .enter_async_context(\n"
" stdio_client(server_params))\n"
" read, write = stdio_transport\n"
" \n"
" # Client session\n"
" self.session = await \\\n"
" self.exit_stack\\\n"
" .enter_async_context(\n"
" ClientSession(read, write))\n"
" \n"
" # Initialize — capability negotiation\n"
" await self.session.initialize()\n"
" \n"
" # List available tools\n"
" response = await \\\n"
" self.session.list_tools()\n"
" tools = response.tools\n"
" print('Connected to server with tools:')\n"
" for t in tools:\n"
" print(f' - {t.name}: '\n"
" f'{t.description}')\n"
" \n"
" return tools\n"
" \n"
" async def process_query(\n"
" self, query: str\n"
" ) -> str:\n"
" '''Process user query with Claude + tools.'''\n"
" # List tools and convert to Claude format\n"
" tools_resp = await \\\n"
" self.session.list_tools()\n"
" claude_tools = [{\n"
" 'name': t.name,\n"
" 'description': t.description,\n"
" 'input_schema': t.inputSchema,\n"
" } for t in tools_resp.tools]\n"
" \n"
" # Send to Claude with tools\n"
" messages = [{'role': 'user',\n"
" 'content': query}]\n"
" response = self.anthropic\\\n"
" .messages.create(\n"
" model='claude-sonnet-4-20250514',\n"
" max_tokens=4096,\n"
" tools=claude_tools,\n"
" messages=messages,\n"
" )\n"
" \n"
" # Process tool calls\n"
" while response.stop_reason == 'tool_use':\n"
" tool_use = next(\n"
" b for b in response.content\n"
" if b.type == 'tool_use')\n"
" tool_name = tool_use.name\n"
" tool_args = tool_use.input\n"
" \n"
" # Execute tool via MCP\n"
" result = await \\\n"
" self.session.call_tool(\n"
" tool_name, tool_args)\n"
" \n"
" # Feed result back to Claude\n"
" messages.append({\n"
" 'role': 'assistant',\n"
" 'content': response.content,\n"
" })\n"
" messages.append({\n"
" 'role': 'user',\n"
" 'content': [{\n"
" 'type': 'tool_result',\n"
" 'tool_use_id':\n"
" tool_use.id,\n"
" 'content': str(\n"
" result.content),\n"
" }],\n"
" })\n"
" \n"
" # Get next response\n"
" response = self.anthropic\\\n"
" .messages.create(\n"
" model=\n"
" 'claude-sonnet-4-20250514',\n"
" max_tokens=4096,\n"
" tools=claude_tools,\n"
" messages=messages,\n"
" )\n"
" \n"
" return response.content[0].text\n"
" \n"
" async def chat_loop(self):\n"
" '''Interactive chat loop.'''\n"
" print('MCP Client started.')\n"
" print('Type queries or quit to exit.')\n"
" \n"
" while True:\n"
" query = input('\\nQuery: ').strip()\n"
" if query.lower() == 'quit':\n"
" break\n"
" response = await \\\n"
" self.process_query(query)\n"
" print(f'\\n{response}')\n"
" \n"
" async def cleanup(self):\n"
" '''Clean up resources.'''\n"
" await self.exit_stack.aclose()\n"
"\n"
"async def main():\n"
" client = MCPClient()\n"
" try:\n"
" await client.connect_to_server(\n"
" 'server.py')\n"
" await client.chat_loop()\n"
" finally:\n"
" await client.cleanup()\n"
"\n"
"if __name__ == '__main__':\n"
" asyncio.run(main())"
)
def get_resource_prompt_usage(self) -> str:
"""Reading resources and using prompts."""
return (
"# Reading resources\n"
"resources = await session\\\n"
" .list_resources()\n"
"for r in resources.resources:\n"
" print(f'URI: {r.uri}, '\n"
" f'Name: {r.name}')\n"
"\n"
"# Read a specific resource\n"
"content = await session.read_resource(\n"
" 'docs://release-checklist')\n"
"print(content.contents[0].text)\n"
"\n"
"# Using prompts\n"
"prompts = await session\\\n"
" .list_prompts()\n"
"for p in prompts.prompts:\n"
" print(f'Name: {p.name}, '\n"
" f'Description: {p.description}')\n"
"\n"
"# Get a prompt with arguments\n"
"result = await session.get_prompt(\n"
" 'code_review_prompt',\n"
" {'owner': 'anthropics',\n"
" 'repo': 'claude-code'})\n"
"print(result.messages[0].content)"
)
def get_capability_negotiation(self) -> dict:
"""Client-side capability negotiation."""
return {
"process": (
"During session.initialize(), client and "
"server exchange capabilities. Client "
"declares sampling, roots, elicitation "
"support. Server declares tools, "
"resources, prompts support."
),
"client_capabilities": [
"Sampling: server can request LLM "
"completions through client",
"Roots: client provides filesystem "
"boundaries to server",
"Elicitation: server can request "
"user input through client",
],
"after_negotiation": (
"Only advertised features can be used. "
"Client knows which tools, resources, "
"and prompts are available. Server "
"knows which client features it can "
"request."
),
}
def get_async_exit_stack(self) -> dict:
"""AsyncExitStack lifecycle management."""
return {
"purpose": (
"Manage lifecycle of stdio_client "
"and ClientSession. Ensures proper "
"cleanup on exit — subprocess "
"terminated, session closed."
),
"pattern": (
"async with AsyncExitStack() as "
"exit_stack:\n"
" read, write = await exit_stack\n"
" .enter_async_context(\n"
" stdio_client(params))\n"
" session = await exit_stack\n"
" .enter_async_context(\n"
" ClientSession(read, write))\n"
" # use session...\n"
"# auto-cleanup on exit"
),
"cleanup": (
"await self.exit_stack.aclose() "
"in finally block ensures cleanup "
"even on errors"
),
}
MCP Client Checklist
- [ ] Install MCP Python SDK: pip install 'mcp>=1.27,<2' and anthropic and python-dotenv
- [ ] Import: from mcp import ClientSession, StdioServerParameters; from mcp.client.stdio import stdio_client
- [ ] Pin mcp>=1.27,<2 — v2 is major rework for 2026-07-28 MCP specification
- [ ] Create StdioServerParameters with command, args, env
- [ ] Use AsyncExitStack for resource lifecycle management
- [ ] Connect via stdio_client(server_params) — spawns server subprocess
- [ ] Create ClientSession(read, write) from stdio transport
- [ ] Call session.initialize() for capability negotiation
- [ ] Client declares: sampling, roots, elicitation support
- [ ] Server declares: tools, resources, prompts support
- [ ] Only advertised features can be used during session
- [ ] List tools: session.list_tools() returns tool schemas (name, description, inputSchema)
- [ ] Call tool: session.call_tool(name, arguments) returns content list
- [ ] List resources: session.list_resources() returns URIs and metadata
- [ ] Read resource: session.read_resource(uri) returns resource content
- [ ] List prompts: session.list_prompts() returns prompt names and schemas
- [ ] Get prompt: session.get_prompt(name, arguments) returns templated message
- [ ] Client discovers capabilities through list calls: tools/list, resources/list, prompts/list
- [ ] No custom JSON-RPC code needed for discovery
- [ ] Integrate with Claude API using Anthropic Python SDK
- [ ] Convert MCP tools to Claude tool format: name, description, input_schema
- [ ] Send messages.create() with tools to Claude
- [ ] Check response.stop_reason for 'tool_use'
- [ ] Extract tool_use block: name and input from response.content
- [ ] Execute tool via session.call_tool(tool_name, tool_args)
- [ ] Feed tool_result back to Claude as user message with tool_result content
- [ ] Loop for multi-turn conversations until stop_reason != 'tool_use'
- [ ] AsyncExitStack ensures proper cleanup: subprocess terminated, session closed
- [ ] Call exit_stack.aclose() in finally block for cleanup on errors
- [ ] stdio transport: client spawns server as subprocess, communicates via stdin/stdout
- [ ] stdio is default for local servers (Claude Desktop, Cursor)
- [ ] Streamable HTTP for remote servers — use mcp.client.streamable_http
- [ ] HTTP+SSE transport deprecated in March 2025 spec update
- [ ] For remote: use streamablehttp_client instead of stdio_client
- [ ] Client can connect to multiple servers — one ClientSession per server
- [ ] Host creates and manages multiple client instances
- [ ] Client maintains isolated session per server — security boundaries
- [ ] Use sys.executable for command to ensure correct Python path
- [ ] Environment variables passed via StdioServerParameters env parameter
- [ ] Async/await throughout — MCP Python SDK is async
- [ ] Tool results are content lists — extract text from content[0].text
- [ ] Resource content is in contents[0].text
- [ ] Prompt result is in messages[0].content
- [ ] Build interactive chat loop for CLI client
- [ ] Handle errors: server not found, tool not found, timeout
- [ ] Read MCP developer guide for protocol
- [ ] Read build MCP server for server side
- [ ] Read Claude Desktop MCP for desktop integration
- [ ] Read MCP transports for transport details
- [ ] Read MCP tools for tool development
- [ ] Test: client connects to server and lists tools correctly
- [ ] Test: tool call returns expected result
- [ ] Test: resource read returns content via URI
- [ ] Test: prompt returns templated message with arguments
- [ ] Test: Claude API integration executes tool calls and feeds results
- [ ] Test: AsyncExitStack cleans up on exit and on errors
- [ ] Test: multi-turn conversation works with tool calls
- [ ] Document client architecture, tool flow, and cleanup strategy
FAQ
How do you build an MCP client in Python?
Use the MCP Python SDK with ClientSession, StdioServerParameters, and stdio_client to connect to an MCP server over stdio transport. MCP Docs: "import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client. Create MCPClient class that connects to server, lists tools, and feeds results to AI." RealPython: "Build an MCP client app for the command line using the MCP Python SDK. Connect to MCP server through stdio transport, list server capabilities, and use server tools to feed an AI-powered chat." MCP Python SDK: "Pin mcp>=1.27,<2 before stable release. v2 is major rework for 2026-07-28 MCP specification." Steps: (1) pip install 'mcp>=1.27,<2'. (2) from mcp import ClientSession, StdioServerParameters. (3) from mcp.client.stdio import stdio_client. (4) Create StdioServerParameters with command and args. (5) Use AsyncExitStack for resource management. (6) Connect via stdio_client, create ClientSession. (7) Call session.initialize() for capability negotiation. (8) List tools with session.list_tools(). (9) Call tools with session.call_tool().
How do you list and call MCP tools from a client?
Use session.list_tools() to discover available tools and session.call_tool() to invoke them with arguments. MCP Docs: "After session.initialize(), call session.list_tools() to get available tools. Use session.call_tool(tool_name, arguments) to invoke a tool." RealPython: "List server capabilities and use server tools to feed an AI-powered chat. Client connects, discovers tools, and the AI model decides which tools to call." Tool listing: tools = await session.list_tools() returns tool schemas with name, description, inputSchema. Tool calling: result = await session.call_tool('get_data', {'query': 'test'}) returns content list. The AI model (Claude) receives tool definitions and decides which to call based on user query. Client executes tool calls and feeds results back to the model.
How do you read resources and use prompts from an MCP client?
Use session.list_resources() and session.read_resource() for resources, session.list_prompts() and session.get_prompt() for prompts. MCP Info: "Client connects to server, discovers capabilities through list calls: tools/list, resources/list, prompts/list." Resources: resources = await session.list_resources() returns URIs and metadata. content = await session.read_resource('docs://checklist') returns resource content. Prompts: prompts = await session.list_prompts() returns prompt names and schemas. result = await session.get_prompt('review_prompt', {'code': '...'}) returns templated message. Resources provide read-only data the AI can access. Prompts provide reusable templates for structuring interactions.
How do you integrate an MCP client with the Claude API?
Use the Anthropic Python SDK to send user messages with available MCP tools to Claude, then execute tool calls through the MCP client and feed results back. MCP Docs: "Create MCPClient class with Anthropic client. Send user message with tools to Claude API. When Claude requests a tool call, execute it through MCP session and feed result back." RealPython: "Use server tools to feed an AI-powered chat. Client connects to MCP server, discovers tools, sends to Claude, executes tool calls, feeds results back." Integration: (1) Create Anthropic client. (2) List MCP tools and convert to Claude tool format. (3) Send messages.create() with tools. (4) Check response for tool_use. (5) Execute tool via session.call_tool(). (6) Feed tool_result back to Claude. (7) Get final response. (8) Loop for multi-turn conversations.
How do you manage MCP client connections with AsyncExitStack?
Use AsyncExitStack from contextlib to manage the lifecycle of stdio_client and ClientSession, ensuring proper cleanup on exit. MCP Docs: "from contextlib import AsyncExitStack. Use AsyncExitStack to manage resource lifecycle. Enter stdio_client context, then ClientSession context. Clean up on exit." MCP Info: "AsyncExitStack manages the connection lifecycle for stdio transport and client session." Pattern: self.session = await self.exit_stack.enter_async_context(stdio_client(server_params)) then self.session = await self.exit_stack.enter_async_context(ClientSession(read, write)). On cleanup: await self.exit_stack.aclose(). This ensures the stdio subprocess and session are properly closed when the client shuts down.
Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →