DEV Community

M Sea Bass
M Sea Bass

Posted on

Practical Guide to MCP (Model Context Protocol) in Python

#mcp #llm

Introduction

MCP (Model Context Protocol) is a protocol that connects LLMs with external tools. This article walks through how to use MCP step by step with real code examples. The full source code is available in the following GitHub repository:
https://github.com/M6saw0/mcp-client-sample

Read time: ~10 minutes

1. Core Concepts of MCP

MCP provides three primary capabilities:

  • Tool: Execute external functionality via function calls
  • Resource: Provide data or information (static or dynamic)
  • Prompt: Provide prompt templates for LLMs

MCP servers mainly support two transport modes. This article covers both stdio and streamable-http:

  • stdio: Communicate via standard I/O (used within the same process)
  • streamable-http: Communicate via HTTP (over the network)

2. Build a Server with stdio

Let’s start by building a server using the simplest stdio transport.

2.1 Basic Server Code 

from mcp.server.fastmcp import FastMCP

# Create a FastMCP instance
mcp = FastMCP(name="FastMCP Demo Server")

# Define a tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

# Define a resource (dynamic)
@mcp.resource("time://{zone}")
def get_time(zone: str) -> str:
    """Return ISO 8601 timestamp."""
    from datetime import datetime, timezone
    now = datetime.now(timezone.utc)
    return now.isoformat() if zone.lower() == "utc" else now.astimezone().isoformat()

# Define a resource (static)
@mcp.resource("info://server")
def get_server_info() -> str:
    """Return server metadata."""
    return "FastMCP demo server"

# Define a prompt
@mcp.prompt()
def greet_user(name: str, tone: str = "friendly") -> str:
    """Generate a greeting instruction."""
    return f"Craft a {tone} greeting addressed to {name}."

# Run the server
if __name__ == "__main__":
    mcp.run(transport="stdio")
Enter fullscreen mode Exit fullscreen mode

2.2 Async Tools and Progress Reporting

For long-running operations, you can report progress:

from typing import Annotated
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.session import ServerSession
import asyncio

@mcp.tool()
async def countdown(
    start: int,
    ctx: Annotated[Context[ServerSession, None], "Injected by FastMCP"],
) -> list[int]:
    """Count down from start to zero."""
    sequence = []
    for step, value in enumerate(range(start, -1, -1), start=1):
        await ctx.report_progress(
            progress=step,
            total=start + 1,
            message=f"Counting value {value}",
        )
        sequence.append(value)
        await asyncio.sleep(0.2)
    return sequence
Enter fullscreen mode Exit fullscreen mode

3. Build a stdio Client

Client code to call the server:

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

async def main():
    server_path = Path(__file__).with_name("server.py")
    server_params = StdioServerParameters(command="python", args=[str(server_path)])

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # List tools
            tools = await session.list_tools()
            print("Available tools:", [tool.name for tool in tools.tools])

            # Call a tool
            result = await session.call_tool("add", arguments={"a": 2, "b": 5})
            print("add result:", result.content)

            # Read a resource
            resource = await session.read_resource("time://local")
            print("time://local:", resource.contents)

            # Get a prompt
            prompt = await session.get_prompt("greet_user", arguments={"name": "Alice"})
            print("prompt:", prompt.messages)

asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

To receive progress reports, pass progress_callback to ClientSession.call_tool:

async def on_progress(progress: float, total: float | None, message: str | None) -> None:
    print(f"{progress}/{total or 0} - {message or ''}")

result = await session.call_tool(
    "countdown",
    arguments={"start": 3},
    progress_callback=on_progress,
)
Enter fullscreen mode Exit fullscreen mode

4. Build a Server with streamable-http

You can convert the stdio server into an HTTP server with just a few changes:

Change: Specify host and port when initializing FastMCP, and use transport="streamable-http".

from mcp.server.fastmcp import FastMCP

# Change: specify host and port
mcp = FastMCP(
    name="FastMCP StreamableHTTP Demo",
    host="127.0.0.1",  # localhost only; 0.0.0.0 also works
    port=8765,
)

# Tool, resource, and prompt definitions are the same
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

# ...(other definitions are the same)

# Change: use transport "streamable-http"
if __name__ == "__main__":
    mcp.run(transport="streamable-http")
Enter fullscreen mode Exit fullscreen mode

5. Build a streamable-http Client

The HTTP client also requires only minimal changes:

Change: Use streamablehttp_client instead of stdio_client and specify a URL.

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

STREAMABLE_HTTP_URL = "http://127.0.0.1:8765/mcp"

async def main():
    # Change: use streamablehttp_client
    async with streamablehttp_client(STREAMABLE_HTTP_URL) as (read, write, get_session_id):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Retrieve the session ID (HTTP-specific)
            if (session_id := get_session_id()) is not None:
                print("Session ID:", session_id)

            # Usage is the same as stdio from here on
            tools = await session.list_tools()
            print("Available tools:", [tool.name for tool in tools.tools])

            result = await session.call_tool("add", arguments={"a": 2, "b": 5})
            print("add result:", result.content)

asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

6. A Client for Multiple Servers

You can handle multiple servers by “keeping and reusing” each ClientSession after establishing connections.

  • Open the transport (stdio/HTTP) → create a ClientSession → call initialize → store in a dictionary for lookup by name 
# Pattern aligned with the repo implementation (async with)
sessions: dict[str, ClientSession] = {}

async with stdio_client(params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        sessions["stdio"] = session  # Keep and reuse if needed
        # You can call call_tool as many times as you want here
        res = await session.call_tool("add", {"a": 1, "b": 2})
        print(res.content)
Enter fullscreen mode Exit fullscreen mode
  • Call call_tool on the retained ClientSession 
result = await sessions[server].call_tool(tool, args)
Enter fullscreen mode Exit fullscreen mode
  • Close everything at the end (centralized lifecycle management for connections and sessions)

This “retain and reuse with centralized lifecycle management” is generalized and organized by MultiServerClient. For implementation details and usage, see multi-server/client.py in the repository. The following functions/methods are the key points:

  • MultiServerClient.__init__: Register servers (stdio/http)
  • connect(): Establish connections to registered servers
  • _ensure_session(name): Check for a session and connect if missing
  • _connect(name): Entry point to create a session (with Future management)
  • _session_task(name, future): Actual connection and lifecycle (async with stdio_client/streamablehttp_client → ClientSession → initialize → wait → close)
  • session(name): Context manager to retrieve a retained ClientSession
  • list_tools(): Enumerate tools from all servers
  • call_tool(server, tool, arguments): Delegate execution to the specified server
  • close(): Safely terminate all sessions (shutdown events and awaiting completion)

7. Using MCP from Generative AI

7.1 Which schema to convert to (Responses vs. Chat Completions)

  • Responses API (e.g., client.responses.create) expects the following format: 
{
  "type": "function",
  "name": "server__tool",
  "description": "...",
  "parameters": { "type": "object", "properties": {"a": {"type": "integer"}}, "required": ["a"], "additionalProperties": false },
  "strict": true
}
Enter fullscreen mode Exit fullscreen mode
  • Chat Completions API (e.g., client.chat.completions.create) expects the following format: 
{
  "type": "function",
  "function": {
    "name": "server__tool",
    "description": "...",
    "parameters": { "type": "object", "properties": {"a": {"type": "integer"}}, "required": ["a"] }
  },
  "strict": true
}
Enter fullscreen mode Exit fullscreen mode

7.2 How to obtain the necessary information (inspect / docstring / MCP meta)

Generate schemas using information obtained from the MCP server via list_tools().

  • Use description and inputSchema
  • In this repository, MultiServerClient.get_tool_details() in for-llm/mcp_client.py aggregates list_tools() results across all servers and provides { "description", "inputSchema" }

You can also generate schemas directly from Python functions.

  • Compose JSON Schema using inspect, type hints, and docstrings
  • The repository’s for-llm/tool_schemas.py implements:
    • function_to_responses_tool(func)
    • function_to_chat_tool(func)
    • build_responses_toolkit(*functions)
    • build_chat_toolkit(*functions)

7.3 Implementation code and concrete methods

See for-llm/run_llm.py in the repository for the actual implementation.

  • Schema conversion utilities: for-llm/tool_schemas.py

    • MCP → Responses: mcp_tool_to_responses_schema(tool_name, description, input_schema, strict=True)
    • MCP → Chat: mcp_tool_to_chat_schema(tool_name, description, input_schema, strict=True)
    • Python function → Responses: function_to_responses_tool(func) / bundle: build_responses_toolkit(...)
    • Python function → Chat: function_to_chat_tool(func) / bundle: build_chat_toolkit(...)

  • Tool metadata retrieval: for-llm/mcp_client.py

    • MultiServerClient.get_tool_details() → { combined_name: {"server_name", "tool_name", "description", "inputSchema"} }

Conclusion

  • MCP provides three capabilities: tools, resources, and prompts
  • stdio communicates within the same process; streamable-http communicates over the network
  • Server and client code are very simple
  • You can centrally manage multiple servers
  • By combining schema conversion with tools, you can dynamically use MCP from generative AI

See the project repository for the full code.

Top comments (0)