4. MCP Clients: Consuming Tools

Overview

In the previous capsule you built an MCP server — a program that exposes tools, resources and prompts through a standard protocol. But a server with nobody using it is like a restaurant with no customers: the food is ready, but nobody ordered it. In this capsule you'll build the other side: the MCP client — the program that connects to the server, discovers which tools it offers, and invokes them.

What makes the MCP client special isn't that it can call remote functions — that already existed with REST APIs. What's special is automatic discovery. Your client connects to a server and asks: "What tools do you have?" The server answers with a complete list — names, descriptions, parameter schemas. Your client didn't need to know in advance which tools existed. It discovered them dynamically, at runtime, without a single line of manual configuration. That's MCP's "wow" moment.

The complete flow you'll implement: connect to the server via stdio, initialize the session, discover tools with list_tools(), invoke tools with call_tool(), access resources with list_resources() and read_resource(), and finally — connect a single client to multiple servers with namespacing to avoid collisions.

The Client-Server Model

Two roles, one protocol

MCP follows a classic client-server model, but with a fundamental difference: the client doesn't need an instruction manual for the server. The protocol lets the client discover the server's capabilities at runtime.

┌─────────────────┐                      ┌─────────────────┐
│   MCP CLIENT    │   1. initialize()    │   MCP SERVER    │
│  (Your agent)   │ ──────────────────→  │  (research_     │
│                 │   2. list_tools()    │   server.py)    │
│                 │ ──────────────────→  │                 │
│                 │   ← tools[]         │                 │
│                 │   3. call_tool()     │                 │
│                 │ ──────────────────→  │  runs the tool  │
│                 │   ← result          │                 │
│                 │   4. list_resources()│                 │
│                 │ ──────────────────→  │                 │
│                 │   ← resources[]     │                 │
└─────────────────┘                      └─────────────────┘
        ▲              Transport                 ▲
        └──────────── (stdio/SSE) ───────────────┘

Each side's responsibilities

The client and the server don't know each other in advance. The only thing they share is the MCP protocol. A client can connect to any MCP server with no custom code. New tools get added to the server → the client discovers them automatically. It's like USB: you don't need a different driver for each keyboard.

Creating an MCP Client

Prerequisites

You need the SDK (pip install mcp) and a server to connect to. We'll use the research_server.py from the previous capsule — the server with 3 tools (search_papers, get_paper_details, summarize_text) and one resource (research://categories). If you don't have it, go back to capsule 03 to build it.

Your first client

The client connects to the server, initializes the session, and it's ready to operate:

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["research_server.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            print("✓ Connected to the MCP server")
            
            tools = await session.list_tools()
            print(f"✓ Available tools: {len(tools.tools)}")
            
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")

asyncio.run(main())

Output:

✓ Connected to the MCP server
✓ Available tools: 3
  - search_papers: Search academic papers by topic
  - get_paper_details: Get the details of a specific paper by ID
  - summarize_text: Summarize a long text into key points

The client's pieces

The nested context manager structure is intentional. The outer one (stdio_client) manages the server process's lifecycle — it launches it on entry, terminates it on exit. The inner one (ClientSession) manages the protocol — it initializes the handshake and closes the session cleanly.

StdioServerParameters in detail

from mcp import StdioServerParameters

params = StdioServerParameters(command="python", args=["research_server.py"])                   # basic
params = StdioServerParameters(command="python", args=["server.py"], cwd="/path/to/project")    # with cwd
params = StdioServerParameters(command="python", args=["server.py"],                            # with env vars
    env={"DATABASE_URL": "postgresql://localhost/papers", "API_KEY": "sk-..."})
params = StdioServerParameters(command="uvx", args=["mcp-server-filesystem", "/home/user"])     # an installed package

The env parameter is useful for passing credentials and configuration without modifying the server's code.

Automatic Tool Discovery

The "wow" moment

This is what makes MCP different from any other tool integration. Your client doesn't know which tools the server has. There's no configuration file, no imports, no hardcoded schemas. The client connects, asks, and discovers.

async def discover_tools():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            tools_response = await session.list_tools()
            
            print(f"I discovered {len(tools_response.tools)} tools:\n")
            for tool in tools_response.tools:
                print(f"Name:        {tool.name}")
                print(f"Description: {tool.description}")
                print(f"Schema:      {tool.inputSchema}")
                print()

asyncio.run(discover_tools())

Stop and think about what just happened. Your client code never mentioned search_papers anywhere. It imported no definitions. It connected to an unknown process and discovered, at runtime, exactly which tools it offers, with which parameters, and what each one does.

The structure of a discovered Tool

Every tool returned by list_tools() has three key fields:

• name: A unique identifier — it's what you pass to call_tool()
• description: Natural language — an LLM uses this to decide whether the tool is relevant
• inputSchema: A JSON Schema of the parameters — it defines arguments, types, and which ones are required

The inputSchema follows the JSON Schema standard, the same format OpenAI and Anthropic use for function calling. Tools discovered via MCP are directly compatible with the schemas LLMs already understand.

Why discovery changes everything

Without MCP, adding a new tool to your agent requires: writing the function, decorating it with @tool, adding it to the list, re-deploying the agent, and configuring credentials.

With MCP: you add the tool to the server, you re-deploy the server. That's it. The client discovers it automatically on the next connection. The agent isn't modified, isn't re-deployed, needs no dependency update.

Calling Remote Tools

call_tool(): invoking by name

Once you've discovered the tools, invoking them is direct:

async def use_tools():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Search papers
            result = await session.call_tool("search_papers", {"query": "RAG techniques", "max_results": 3})
            print("=== Search ===")
            for content in result.content:
                print(content.text)
            
            # Get the details
            result = await session.call_tool("get_paper_details", {"paper_id": "arxiv-2312.10997"})
            print("\n=== Details ===")
            for content in result.content:
                print(content.text)
            
            # Summarize
            result = await session.call_tool(
                "summarize_text",
                {"text": "RAG combines retrieval with generation...", "style": "bullets"}
            )
            print("\n=== Summary ===")
            for content in result.content:
                print(content.text)

asyncio.run(use_tools())

The result and error handling

call_tool() returns a CallToolResult with content (text, image, or an embedded resource) and an isError flag. Remote tools can fail, so your client needs a robust wrapper:

import asyncio

async def safe_call_tool(session: ClientSession, tool_name: str, arguments: dict):
    """A wrapper with a timeout and error handling."""
    try:
        result = await asyncio.wait_for(session.call_tool(tool_name, arguments), timeout=30.0)
        
        if result.isError:
            print(f"[ERROR] '{tool_name}': {result.content[0].text}")
            return None
        return result
    
    except asyncio.TimeoutError:
        print(f"[TIMEOUT] '{tool_name}' didn't respond in 30s")
        return None
    except Exception as e:
        print(f"[ERROR] '{tool_name}': {e}")
        return None

Pattern: discovery + dynamic invocation

The most powerful pattern combines discovery with invocation. The client discovers the tools and builds a map to invoke them dynamically:

async def dynamic_client():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Discovery → a tool map
            tools = await session.list_tools()
            tool_map = {tool.name: tool for tool in tools.tools}
            
            # Dynamic selection (in an agent, the LLM chooses)
            selected = "search_papers"
            schema = tool_map[selected].inputSchema
            required = schema.get("required", [])
            
            print(f"Tool: {selected}")
            print(f"Required: {required}")
            
            result = await session.call_tool(selected, {"query": "transformers"})
            for content in result.content:
                print(content.text)

asyncio.run(dynamic_client())

Accessing Resources

Tools vs Resources

Discovering and reading resources

The pattern is identical to tools — first you discover, then you access:

async def explore_resources():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Discover resources
            resources = await session.list_resources()
            print(f"Available resources: {len(resources.resources)}\n")
            for resource in resources.resources:
                print(f"  URI:  {resource.uri}")
                print(f"  Name: {resource.name}")
                print(f"  MIME: {resource.mimeType}\n")
            
            # Read the content
            result = await session.read_resource("research://categories")
            for content in result.contents:
                print(f"Content: {content.text}")

asyncio.run(explore_resources())

A common pattern: read resources to get context, then use tools with more precise information:

# Get the available categories → search within a valid category
categories = await session.read_resource("research://categories")
available = categories.contents[0].text  # "RAG, Fine-tuning, ..."

result = await session.call_tool("search_papers", {"query": "RAG", "max_results": 3})

Multi-Server Connections

One agent, many tool sources

In production, your agent may need a filesystem server, a web search server, and a database server. Each one is an independent process. What happens if two servers have a search tool? Without namespacing, you get a collision. The solution: prefix each tool with the server's name.

class MultiServerClient:
    def __init__(self):
        self.servers: dict[str, StdioServerParameters] = {}
        self.tool_registry: dict[str, dict] = {}
    
    def add_server(self, name: str, params: StdioServerParameters):
        self.servers[name] = params
    
    async def discover_all_tools(self):
        self.tool_registry = {}
        for server_name, params in self.servers.items():
            async with stdio_client(params) as (read, write):
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    tools = await session.list_tools()
                    for tool in tools.tools:
                        namespaced = f"{server_name}__{tool.name}"
                        self.tool_registry[namespaced] = {
                            "server": server_name, "original_name": tool.name,
                            "description": f"[{server_name}] {tool.description}",
                            "schema": tool.inputSchema,
                        }
        return self.tool_registry
    
    async def call_tool(self, namespaced_name: str, arguments: dict):
        entry = self.tool_registry[namespaced_name]
        params = self.servers[entry["server"]]
        async with stdio_client(params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                return await session.call_tool(entry["original_name"], arguments)

# Usage
async def demo_multi_server():
    client = MultiServerClient()
    client.add_server("research", StdioServerParameters(command="python", args=["research_server.py"]))
    client.add_server("filesystem", StdioServerParameters(command="python", args=["filesystem_server.py"]))
    
    await client.discover_all_tools()
    for name, info in client.tool_registry.items():
        print(f"  {name}: {info['description']}")
    # research__search_papers: [research] Search academic papers
    # filesystem__read_file: [filesystem] Read a file
    
    result = await client.call_tool("research__search_papers", {"query": "MCP"})

The server__tool convention (a double underscore) is clear and easy to parse. What matters is being consistent. In this example, each call_tool opens and closes a connection; in production you'll want a pool of persistent connections (capsule 07).

Connection to the Project

Module 7 — Research Agent with MCP

In capsule 08 you'll integrate everything from this module into the Research Agent:

• Discovery at startup: When the Research Agent boots, it connects to its MCP servers and discovers every available tool, without knowing in advance which servers have which tools
• call_tool() as the invocation: Every time the agent decides to use a tool, it calls it via session.call_tool() — the server runs the logic, not the agent
• Resources for context: The agent reads the server's resources to get metadata (categories, configuration) before deciding which tools to invoke
• Multi-server: The Research Agent connects to 3 servers (filesystem, web search, paper database), each discovered dynamically

The next capsules in this module

Troubleshooting

Problem 1: "Connection refused" or the client hangs on connect

Symptom: When trying stdio_client(server_params), you get an error or the program doesn't advance.

Cause: The command or args in StdioServerParameters are wrong. The server doesn't launch.

Solution: Verify the server runs manually first (python research_server.py). Use an absolute path or an explicit cwd:

# INCORRECT: a relative path can fail
params = StdioServerParameters(command="python", args=["server.py"])

# CORRECT: an explicit path
params = StdioServerParameters(command="python", args=["research_server.py"], cwd="/path/to/project")

Problem 2: list_tools() returns an empty list

Symptom: The connection succeeds, but zero tools get discovered.

Cause: The server doesn't have @app.list_tools() or the handler returns [].

Solution: Add logging to the server (use sys.stderr because stdout is reserved for MCP in stdio):

@app.list_tools()
async def list_tools() -> list[Tool]:
    tools = [Tool(name="search_papers", ...)]
    import sys
    print(f"[SERVER] Returning {len(tools)} tools", file=sys.stderr)
    return tools

Problem 3: call_tool() fails with "Tool not found"

Symptom: The tool shows up in list_tools(), but call_tool() errors.

Cause: The name doesn't match exactly (capitalization, an extra space).

Solution: Use the exact names from discovery:

tools = await session.list_tools()
valid_names = {tool.name for tool in tools.tools}

if tool_to_call not in valid_names:
    print(f"'{tool_to_call}' doesn't exist. Valid: {valid_names}")

Problem 4: The client hangs after initialize()

Symptom: initialize() passes, but list_tools() or call_tool() wait forever.

Cause: The server terminated prematurely but the stdio connection is still open.

Solution: Add timeouts:

try:
    tools = await asyncio.wait_for(session.list_tools(), timeout=10.0)
except asyncio.TimeoutError:
    print("The server didn't respond. Check that it isn't terminating prematurely.")

Problem 5: Serialization errors in the arguments

Symptom: call_tool() fails because the arguments don't match the schema.

Cause: A wrong type or a missing required field.

Solution: Validate against the schema before sending. Use tool.inputSchema["required"] to check the required fields and tool.inputSchema["properties"] to check types and valid fields.

Exercises

Exercise 1: A basic client with a tool report (Easy)

Create an MCP client that connects to research_server.py, discovers every tool, and prints a formatted report: name, description, required parameters, and optional parameters.

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def tool_report():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    # Connect, discover tools, and print the report
    # For each tool show: name, description, required, optional
    # Your code here...

asyncio.run(tool_report())

async def tool_report():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    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(f"=== Tool Report ({len(tools.tools)} found) ===\n")
            
            for tool in tools.tools:
                schema = tool.inputSchema
                properties = schema.get("properties", {})
                required = set(schema.get("required", []))
                
                req_params = [f"{p} ({properties[p].get('type', 'any')})" for p in required]
                opt_params = [
                    f"{p} ({properties[p].get('type', 'any')}, default={properties[p].get('default', 'N/A')})"
                    for p in properties if p not in required
                ]
                
                print(f"Tool: {tool.name}")
                print(f"  Description: {tool.description}")
                print(f"  Required:    {', '.join(req_params) if req_params else 'none'}")
                print(f"  Optional:    {', '.join(opt_params) if opt_params else 'none'}")
                print()

asyncio.run(tool_report())

Exercise 2: Chaining tools (Medium)

Connect to the server, search for papers with search_papers, take the result, and pass it to summarize_text. The final output should be a summary of the papers it found.

async def chained_tools():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    # 1. Search papers about "transformer architectures"
    # 2. Take the result's text
    # 3. Pass it to summarize_text with style="bullets"
    # 4. Print the final summary
    # Your code here...

asyncio.run(chained_tools())

async def chained_tools():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            search_result = await session.call_tool(
                "search_papers", {"query": "transformer architectures", "max_results": 3}
            )
            search_text = search_result.content[0].text
            print(f"=== Search ===\n{search_text}\n")
            
            summary_result = await session.call_tool(
                "summarize_text", {"text": search_text, "style": "bullets"}
            )
            print(f"=== Summary ===\n{summary_result.content[0].text}")

asyncio.run(chained_tools())

Exercise 3: Resource-informed tool calling (Medium)

Read the research://categories resource to get the available categories. Then search for papers in each category and show a report with the number of results per category.

async def informed_search():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    # 1. Read research://categories
    # 2. Parse the categories (comma-separated)
    # 3. For each category, call search_papers
    # 4. Print a table: category → number of results
    # Your code here...

asyncio.run(informed_search())

async def informed_search():
    server_params = StdioServerParameters(command="python", args=["research_server.py"])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            resource_result = await session.read_resource("research://categories")
            categories = [c.strip() for c in resource_result.contents[0].text.split(",")]
            print(f"Categories discovered: {categories}\n")
            
            print(f"{'Category':<25} {'Results':<10}")
            print("-" * 35)
            
            for category in categories:
                result = await session.call_tool("search_papers", {"query": category, "max_results": 5})
                n_results = len(result.content[0].text.strip().split("\n"))
                print(f"{category:<25} {n_results:<10}")

asyncio.run(informed_search())

Exercise 4: A client with schema validation (Hard)

Implement validated_call_tool that validates the arguments against the inputSchema before invoking: required fields present, no unknown fields, correct basic types. If it fails, return (None, errors) without making the call; if it passes, return (result, None).

async def validated_call_tool(session, tool_name: str, arguments: dict):
    # 1. Get the tool and its schema via list_tools()
    # 2. Validate required fields, unknown fields, and types
    # 3. If it passes, call call_tool()
    # Your code here...
    pass

TYPE_VALIDATORS = {
    "string": lambda v: isinstance(v, str),
    "integer": lambda v: isinstance(v, int) and not isinstance(v, bool),
    "number": lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
    "boolean": lambda v: isinstance(v, bool),
}

async def validated_call_tool(session: ClientSession, tool_name: str, arguments: dict):
    tools = await session.list_tools()
    tool = next((t for t in tools.tools if t.name == tool_name), None)
    
    if not tool:
        return None, [f"Tool '{tool_name}' doesn't exist"]
    
    props = tool.inputSchema.get("properties", {})
    required = set(tool.inputSchema.get("required", []))
    errors = []
    
    for f in required:
        if f not in arguments:
            errors.append(f"Missing required field: '{f}'")
    for f in arguments:
        if f not in props:
            errors.append(f"Unknown field: '{f}'")
    for f, v in arguments.items():
        if f in props:
            expected = props[f].get("type")
            if expected in TYPE_VALIDATORS and not TYPE_VALIDATORS[expected](v):
                errors.append(f"Wrong type for '{f}': expected {expected}, got {type(v).__name__}")
    
    if errors:
        return None, errors
    return await session.call_tool(tool_name, arguments), None

# Test
async def test():
    params = StdioServerParameters(command="python", args=["research_server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            print(await validated_call_tool(session, "search_papers", {"query": "RAG"}))       # OK
            print(await validated_call_tool(session, "search_papers", {"max_results": 5}))     # query missing
            print(await validated_call_tool(session, "search_papers", {"query": 42}))          # wrong type

Exercise 5: Unified multi-server discovery (Hard)

Implement a UnifiedToolRegistry: connect to multiple servers, register tools with a namespace (server__tool), and allow searching by keyword via search_tools(keyword).

class UnifiedToolRegistry:
    def __init__(self): ...
    def add_server(self, name, params): ...
    async def discover_all(self): ...
    def search_tools(self, keyword) -> list[dict]: ...

class UnifiedToolRegistry:
    def __init__(self):
        self.servers: dict[str, StdioServerParameters] = {}
        self.tools: dict[str, dict] = {}
    
    def add_server(self, name: str, params: StdioServerParameters):
        self.servers[name] = params
    
    async def discover_all(self):
        self.tools = {}
        for name, params in self.servers.items():
            try:
                async with stdio_client(params) as (read, write):
                    async with ClientSession(read, write) as session:
                        await session.initialize()
                        for tool in (await session.list_tools()).tools:
                            key = f"{name}__{tool.name}"
                            self.tools[key] = {
                                "server": name, "description": tool.description or "",
                                "params": list(tool.inputSchema.get("properties", {}).keys()),
                            }
                        print(f"[{name}] {len(self.tools)} tools")
            except Exception as e:
                print(f"[{name}] Error: {e}")
    
    def search_tools(self, keyword: str) -> list[dict]:
        kw = keyword.lower()
        return [{"name": k, **v} for k, v in self.tools.items()
                if kw in k.lower() or kw in v["description"].lower()]

async def demo():
    registry = UnifiedToolRegistry()
    registry.add_server("research", StdioServerParameters(command="python", args=["research_server.py"]))
    await registry.discover_all()
    print(f"Tools with 'paper': {[r['name'] for r in registry.search_tools('paper')]}")

Summary

In this capsule you learned:

• MCP's client-server model: the client connects, discovers, and invokes. The server exposes, executes, and responds. They don't know each other in advance — the protocol makes the interaction possible with no prior configuration
• Creating an MCP client: three pieces — StdioServerParameters to define how to launch the server, stdio_client to open the connection, and ClientSession to handle the protocol
• Automatic tool discovery: list_tools() returns every tool from the server with its name, description, and schema. Your client discovers tools it didn't know about, at runtime, with no manual configuration. This is MCP's "wow" moment
• Invoking remote tools: call_tool(name, arguments) runs the tool on the server and returns the result. Error handling with result.isError and wrappers with timeouts
• Accessing resources: list_resources() and read_resource(uri) to discover and read the server's data, which gives context for invoking tools more intelligently
• Multi-server connections: one client can connect to multiple servers. Namespacing (server__tool) avoids collisions. A unified registry lets you search tools across every server

Next capsule: MCP in LangGraph Agents. You'll take the tools discovered via MCP and turn them into LangChain tools a LangGraph agent can use inside its StateGraph — dynamic tool loading where the LLM chooses which MCP tool to invoke.

Additional Resources

1. MCP Client Concepts — Official documentation on the MCP client's architecture, capabilities, and lifecycle
2. MCP Python SDK — The SDK's source code and examples, including ClientSession and transports
3. MCP Specification — Tools — The formal protocol specification for list_tools and call_tool
4. MCP Specification — Resources — The formal specification for resources and URIs
5. Building MCP Clients (Tutorial) — A step-by-step tutorial for building clients that connect to servers