Model Context Protocol (MCP)

What Is MCP?

Model Context Protocol (MCP) is an open standard introduced by Anthropic in late 2024 and rapidly adopted by OpenAI, Google, and the entire ecosystem. It defines a standard protocol for AI agents to communicate with external tools and data sources.

flowchart LR
    subgraph Host["MCP Host"]
        AG["Your Agent\nClaude / LangGraph / any app"]
        CL["MCP Client\nmanages server lifecycle"]
    end
    subgraph Server["MCP Server\nGitHub · Postgres · Slack · custom"]
        T["Tools\nactions"]
        R["Resources\ndata"]
        P["Prompts\ntemplates"]
    end
    AG --- CL
    CL <-->|"MCP protocol\nstdio or HTTP+SSE"| Server
    style AG fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style CL fill:#EEF0F7,stroke:#818CF8,color:#0F172A
    style T fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style R fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style P fill:#f0fdf9,stroke:#0D9488,color:#0F172A

The key insight: once you build an MCP server, it works with any MCP-compatible agent without modification.

Building an MCP Server

# src/mcp_server/server.py
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server   # stdio transport for local subprocess communication
from mcp import types
import httpx
import json
from datetime import datetime

# Server("company-data-server") creates the MCP server with a name that MCP clients use to identify it
server = Server("company-data-server")

# ─── TOOLS ──────────────────────────────────────────────────────

# @server.list_tools() decorator registers this as the tool discovery handler: MCP clients call it first
# to learn what actions are available before asking the agent to do anything
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """Return available tools to the MCP client."""
    return [
        types.Tool(
            name="search_products",
            description="Search the product catalog by name, category, or SKU",
            # inputSchema is a JSON Schema object: MCP clients use it to validate arguments
            # and LLMs use the description to decide when and how to call this tool
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query (product name, category, or SKU)"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Maximum results to return (1-50)",
                        "default": 10,
                    },
                    "in_stock_only": {
                        "type": "boolean",
                        "description": "Filter to in-stock items only",
                        "default": False,
                    }
                },
                "required": ["query"],   # only query is required; limit and in_stock_only have defaults
            }
        ),
        types.Tool(
            name="get_customer",
            description="Retrieve customer information by email or customer ID",
            inputSchema={
                "type": "object",
                "properties": {
                    "identifier": {"type": "string", "description": "Customer email or ID"},
                },
                "required": ["identifier"],
            }
        ),
        types.Tool(
            name="create_support_ticket",
            description="Create a customer support ticket. Use when the issue cannot be resolved immediately.",
            inputSchema={
                "type": "object",
                "properties": {
                    "customer_email": {"type": "string"},
                    "subject": {"type": "string"},
                    "description": {"type": "string"},
                    "priority": {"type": "string", "enum": ["low", "medium", "high"]},  # enum restricts valid values
                },
                "required": ["customer_email", "subject", "description"],
            }
        ),
    ]

# @server.call_tool() handles all tool invocations: routes by name to the right implementation
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Execute a tool call and return results."""
    
    if name == "search_products":
        results = await search_product_database(
            query=arguments["query"],
            limit=arguments.get("limit", 10),            # .get() with default handles optional args safely
            in_stock_only=arguments.get("in_stock_only", False),
        )
        # MCP tool results must be returned as a list of content objects; TextContent wraps JSON strings
        return [types.TextContent(type="text", text=json.dumps(results, indent=2))]
    
    elif name == "get_customer":
        customer = await get_customer_from_db(arguments["identifier"])
        if not customer:
            # return a structured error in the content rather than raising: lets the LLM handle "not found"
            return [types.TextContent(type="text", text=json.dumps({"error": "Customer not found"}))]
        return [types.TextContent(type="text", text=json.dumps(customer, indent=2))]
    
    elif name == "create_support_ticket":
        ticket_id = await create_ticket_in_crm(
            email=arguments["customer_email"],
            subject=arguments["subject"],
            description=arguments["description"],
            priority=arguments.get("priority", "medium"),  # default to medium if priority not specified
        )
        return [types.TextContent(type="text", text=json.dumps({"ticket_id": ticket_id, "status": "created"}))]
    
    else:
        # raise for truly unknown tools: this is a server error, not a user error
        raise ValueError(f"Unknown tool: {name}")

# ─── RESOURCES ──────────────────────────────────────────────────

# @server.list_resources() registers the resource discovery handler: resources are read-only context
# the model can read but not modify, unlike tools which perform actions
@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="company://policy/return-policy",   # custom URI scheme identifies where the resource lives
            name="Return Policy",
            description="The company's official return and refund policy",
            mimeType="text/plain",
        ),
        types.Resource(
            uri="company://catalog/categories",
            name="Product Categories",
            description="Full list of product categories and subcategories",
            mimeType="application/json",
        ),
    ]

# @server.read_resource() is called when the MCP client requests a specific resource by URI
@server.read_resource()
async def handle_read_resource(uri: str) -> str:
    if uri == "company://policy/return-policy":
        return "Our return policy: Items can be returned within 30 days..."
    elif uri == "company://catalog/categories":
        categories = await get_all_categories()
        return json.dumps(categories)  # return JSON string; mimeType="application/json" hints how to parse it
    raise ValueError(f"Resource not found: {uri}")

# ─── RUN SERVER ─────────────────────────────────────────────────

async def main():
    # stdio_server() manages stdin/stdout as bidirectional streams for local MCP communication
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="company-data-server",
                server_version="1.0.0",   # version is surfaced to the MCP client for compatibility checks
            )
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

HTTP Transport for Network MCP Servers

For MCP servers that need to be accessed over the network:

# src/mcp_server/http_server.py
from mcp.server.sse import SseServerTransport   # SSE transport for network MCP, supports auth headers and load balancers
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI(title="Company Data MCP Server")
# SseServerTransport("/messages"): "/messages" is the POST endpoint path for client-to-server messages
sse_transport = SseServerTransport("/messages")

@app.get("/sse")
async def sse_endpoint(request: Request):
    """SSE endpoint for MCP client connections."""
    # connect_sse opens a persistent SSE connection; server.run handles the MCP protocol over it
    async with sse_transport.connect_sse(request.scope, request.receive, request._send) as streams:
        await server.run(*streams, InitializationOptions(
            server_name="company-data-server",
            server_version="1.0.0",
        ))

@app.post("/messages")
async def messages_endpoint(request: Request):
    # MCP clients POST tool calls and resource requests here; SSE sends responses back on the /sse stream
    await sse_transport.handle_post_message(request.scope, request.receive, request._send)

Consuming MCP Servers in LangGraph

# src/agents/mcp_agent.py
from langchain_mcp_adapters.client import MultiServerMCPClient   # bridges MCP servers to LangChain tools

async def create_mcp_agent():
    """Create an agent that uses tools from MCP servers."""
    
    # MultiServerMCPClient connects to multiple MCP servers simultaneously
    # each key is a logical name; the dict value specifies transport and connection details
    client = MultiServerMCPClient(
        {
            "company-data": {
                "command": "python",                          # launch as a local subprocess
                "args": ["src/mcp_server/server.py"],
                "transport": "stdio",                         # communicate via stdin/stdout
            },
            "github": {
                "url": "https://mcp-github.example.com/sse",
                "transport": "sse",                           # network transport, works across machines
                "headers": {"Authorization": f"Bearer {settings.github_mcp_token}"},  # auth header for the remote server
            },
            "filesystem": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"],  # official MCP filesystem server
                "transport": "stdio",
            }
        }
    )
    
    # get_tools() discovers all tools from all connected servers and wraps them as LangChain BaseTool objects
    tools = await client.get_tools()
    
    print(f"Loaded {len(tools)} tools from MCP servers:")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description[:60]}")
    
    # create_react_agent accepts the MCP-sourced tools exactly like hand-written LangChain tools
    agent = create_react_agent(
        model=ChatOpenAI(model="gpt-5.6-sol"),
        tools=tools,   # all tools from all MCP servers are available to the agent
    )
    
    return agent, client

# Usage
async def main():
    agent, mcp_client = await create_mcp_agent()
    
    # async with mcp_client ensures MCP subprocess connections are cleaned up on exit
    async with mcp_client:
        result = await agent.ainvoke({
            "messages": [HumanMessage(content="Search for all MacBook products in stock")]
        })
        
        print(result["messages"][-1].content)

MCP Server Configuration (mcp.json)

Projects that consume multiple MCP servers use a configuration file:

{
  "mcpServers": {
    "company-data": {
      "command": "python",
      "args": ["src/mcp_server/server.py"],
      "env": {
        "DATABASE_URL": "${DATABASE_URL}"   // env var substitution, secrets stay in shell environment
      }
    },
    "brave-search": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-brave-search"],   // -y auto-accepts npx install prompt
      "env": {
        "BRAVE_API_KEY": "${BRAVE_API_KEY}"
      }
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "${DATABASE_URL}"]   // DATABASE_URL passed as a CLI arg
    }
  }
}

MCP is still evolving rapidly, new MCP servers for every major platform (Slack, Notion, Jira, Salesforce, AWS) are being published weekly. Your agent can consume all of them through the same standard protocol, without writing any provider-specific integration code.