LangChain SDK Track Part 7: MCP — Model Context Protocol Integration

1. MCP Overview

The Model Context Protocol (MCP) is an open standard that defines how AI applications connect to external tools, data sources, and services. Think of it as a USB for AI — a universal protocol that lets any AI model interact with any tool through a standardized interface, regardless of the model provider or the tool implementation.

LangChain integrates with MCP via the langchain-mcp-adapters library, which converts MCP tools into LangChain tools — making them directly usable in any LangChain agent, LangGraph workflow, or Deep Agent.

1.1 MCP Architecture

flowchart LR
    A["MCP Host
(Your App / Agent)"] --> B["MCP Client
(langchain-mcp-adapters)"]
    B -->|"stdio"| C["MCP Server A
(Local Process)"]
    B -->|"HTTP"| D["MCP Server B
(Remote Service)"]
    C --> E["Tools: add, multiply"]
    D --> F["Tools: get_weather, search"]
                        

1.2 Why MCP Matters

2. Building MCP Servers

MCP servers expose tools (callable functions), resources (readable data), and prompts (reusable templates) that any MCP-compatible client can discover and use. The FastMCP library is the recommended way to build custom MCP servers.

2.1 FastMCP Setup

# Install FastMCP
pip install fastmcp

# Or with uv (faster)
uv add fastmcp

2.2 Defining Tools with FastMCP

A minimal MCP server with two tools — one for math, one for weather. Each @mcp.tool() decorated function becomes a tool discoverable by any MCP client:

from fastmcp import FastMCP

# Create a named MCP server
mcp = FastMCP("Math")

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

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    # Run with stdio transport (for local subprocess communication)
    mcp.run(transport="stdio")

A separate weather server using HTTP transport for remote access:

from fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """Get weather for a given location.

    Args:
        location: City name or coordinates
    """
    # In production, call a real weather API here
    return f"Weather in {location}: 22°C, Partly Cloudy"

@mcp.tool()
async def get_forecast(location: str, days: int = 5) -> str:
    """Get multi-day weather forecast.

    Args:
        location: City name or coordinates
        days: Number of forecast days (1-10)
    """
    return f"{days}-day forecast for {location}: Sunny, 20-25°C"

if __name__ == "__main__":
    # Run with HTTP transport (Streamable HTTP for remote access)
    mcp.run(transport="streamable-http", port=8000)

2.3 Resources & Prompts

MCP servers can also expose resources (readable data) and prompts (reusable templates) alongside tools:

from fastmcp import FastMCP

mcp = FastMCP("KnowledgeBase")

@mcp.tool()
def search_docs(query: str, max_results: int = 5) -> str:
    """Search the documentation knowledge base.

    Args:
        query: Search query string
        max_results: Maximum number of results to return
    """
    # Replace with actual search logic
    return f"Found {max_results} results for '{query}'"

@mcp.resource("docs://api-reference")
def get_api_reference() -> str:
    """Return the current API reference documentation."""
    return "# API Reference\n\n## Endpoints\n- GET /users\n- POST /orders"

@mcp.prompt("summarize")
def summarize_prompt(text: str) -> str:
    """A reusable prompt for summarizing text."""
    return f"Please summarize the following text concisely:\n\n{text}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http", port=8001)

3. Connecting LangChain Agents to MCP Servers

The langchain-mcp-adapters library converts MCP tools into LangChain tools, making them usable in any agent or workflow. The core class is MultiServerMCPClient, which manages connections to one or more MCP servers.

3.1 MultiServerMCPClient

# Install the adapter library
pip install langchain-mcp-adapters

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def main():
    # Connect to multiple MCP servers simultaneously
    client = MultiServerMCPClient(
        {
            "math": {
                "transport": "stdio",
                "command": "python",
                "args": ["/path/to/math_server.py"],
            },
            "weather": {
                "transport": "http",
                "url": "http://localhost:8000/mcp",
            }
        }
    )

    # Get all tools from all connected servers
    tools = await client.get_tools()
    print(f"Discovered {len(tools)} tools from MCP servers")

    # Create a LangChain agent with MCP tools
    agent = create_agent("openai:gpt-4o-mini", tools)

    # Invoke — the agent automatically selects the right MCP tool
    math_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "What's (3 + 5) x 12?"}]}
    )
    print(math_response["messages"][-1].content)

    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "What's the weather in NYC?"}]}
    )
    print(weather_response["messages"][-1].content)

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

3.2 Transport Mechanisms

MCP supports different transport types for client-server communication:

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient

async def demo_transports():
    client = MultiServerMCPClient(
        {
            # stdio: launches server as a subprocess
            "local_tools": {
                "transport": "stdio",
                "command": "python",
                "args": ["./servers/local_tools.py"],
            },
            # HTTP with custom headers (auth, tracing)
            "remote_api": {
                "transport": "http",
                "url": "http://api.example.com/mcp",
                "headers": {
                    "Authorization": "Bearer YOUR_TOKEN",
                    "X-Trace-Id": "session-001",
                },
            },
        }
    )

    tools = await client.get_tools()
    print(f"Total tools available: {len(tools)}")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")

asyncio.run(demo_transports())

3.3 Stateful Sessions

When working with MCP servers that maintain state across tool calls (e.g., a database connection that keeps a transaction open), create a persistent session:

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain.agents import create_agent

async def stateful_example():
    client = MultiServerMCPClient(
        {
            "database": {
                "transport": "http",
                "url": "http://localhost:9000/mcp",
            }
        }
    )

    # Create an explicit session — persists across tool calls
    async with client.session("database") as session:
        # Load tools within the session context
        tools = await load_mcp_tools(session)

        agent = create_agent("openai:gpt-4o-mini", tools)

        # Both calls share the same MCP session
        # (e.g., same DB transaction, same auth context)
        result1 = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "Begin a transaction and insert a user named Alice"}]}
        )
        result2 = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "Now commit the transaction"}]}
        )
        print(result2["messages"][-1].content)

asyncio.run(stateful_example())

4. Advanced Patterns

4.1 Structured Content

MCP tools can return structured content (JSON) alongside human-readable text. The adapter wraps structured content in an MCPToolArtifact accessible via the artifact field on the ToolMessage:

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_core.messages import ToolMessage

async def structured_content_example():
    client = MultiServerMCPClient(
        {"data_api": {"transport": "http", "url": "http://localhost:8000/mcp"}}
    )
    tools = await client.get_tools()
    agent = create_agent("openai:gpt-4o-mini", tools)

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Get the sales data for Q1 2026"}]}
    )

    # Extract structured content from tool messages
    for message in result["messages"]:
        if isinstance(message, ToolMessage) and message.artifact:
            structured_data = message.artifact["structured_content"]
            print(f"Structured response: {structured_data}")

asyncio.run(structured_content_example())

4.2 Authentication

For production MCP servers requiring authentication, pass headers or implement the httpx.Auth interface:

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def authenticated_mcp():
    client = MultiServerMCPClient(
        {
            "secure_api": {
                "transport": "http",
                "url": "https://api.company.com/mcp",
                "headers": {
                    "Authorization": "Bearer eyJhbGciOiJSUzI1NiIs...",
                    "X-API-Version": "2026-05",
                },
            }
        }
    )

    tools = await client.get_tools()
    agent = create_agent("anthropic:claude-sonnet-4-6", tools)

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "List my recent deployments"}]}
    )
    print(result["messages"][-1].content)

asyncio.run(authenticated_mcp())

4.3 Interceptors

Interceptors let you transform tool inputs/outputs before they reach the agent. Use them for logging, validation, or appending structured content to the conversation:

import json
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain.agents import create_agent
from mcp.types import TextContent

async def log_and_transform(request: MCPToolCallRequest, handler):
    """Interceptor that logs tool calls and appends structured content."""
    print(f"[MCP] Calling tool: {request.tool_name} with args: {request.arguments}")

    # Execute the tool
    result = await handler(request)

    # If the tool returned structured content, append it as visible text
    if result.structuredContent:
        result.content += [
            TextContent(type="text", text=json.dumps(result.structuredContent, indent=2))
        ]

    print(f"[MCP] Tool {request.tool_name} completed")
    return result

async def interceptor_example():
    # Pass interceptors at client creation
    client = MultiServerMCPClient(
        {"api": {"transport": "http", "url": "http://localhost:8000/mcp"}},
        tool_interceptors=[log_and_transform]
    )

    tools = await client.get_tools()
    agent = create_agent("openai:gpt-4o-mini", tools)

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Search for 'rate limiting best practices'"}]}
    )
    print(result["messages"][-1].content)

asyncio.run(interceptor_example())

5. LangGraph + MCP

5.1 MCP Tools in LangGraph Workflows

MCP tools integrate seamlessly with LangGraph StateGraph workflows. Load tools from MCP servers and pass them to nodes that need external capabilities:

import asyncio
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

class State(TypedDict):
    messages: Annotated[list, add_messages]

async def build_mcp_workflow():
    # Load MCP tools
    client = MultiServerMCPClient(
        {
            "search": {"transport": "http", "url": "http://localhost:8000/mcp"},
            "database": {"transport": "http", "url": "http://localhost:8001/mcp"},
        }
    )
    tools = await client.get_tools()

    # Bind tools to model
    model = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

    def agent_node(state: State) -> dict:
        response = model.invoke(state["messages"])
        return {"messages": [response]}

    def should_use_tools(state: State) -> str:
        last_msg = state["messages"][-1]
        if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
            return "tools"
        return END

    # Build the graph
    graph = StateGraph(State)
    graph.add_node("agent", agent_node)
    graph.add_node("tools", ToolNode(tools))

    graph.add_edge(START, "agent")
    graph.add_conditional_edges("agent", should_use_tools, {"tools": "tools", END: END})
    graph.add_edge("tools", "agent")

    app = graph.compile()

    # Run the workflow
    result = await app.ainvoke(
        {"messages": [HumanMessage(content="Search for Python best practices and save to database")]}
    )
    print(result["messages"][-1].content)

asyncio.run(build_mcp_workflow())

5.2 Exposing LangGraph Agents as MCP Servers

Agent Server implements MCP using the Streamable HTTP transport, allowing your LangGraph agents to be exposed as MCP tools themselves — making them usable by any MCP-compliant client:

flowchart TB
    A["External MCP Client
(Claude Desktop, IDE)"] -->|"/mcp endpoint"| B["Agent Server
(Your LangGraph Agent)"]
    B -->|"MCP Client"| C["MCP Server A
(Database)"]
    B -->|"MCP Client"| D["MCP Server B
(Search)"]
    B -->|"MCP Client"| E["MCP Server C
(GitHub)"]
                        

6. Production Patterns

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from deepagents import create_deep_agent

async def production_mcp_agent():
    """Production pattern: Deep Agent with MCP tools from multiple servers."""

    # Connect to production MCP servers
    client = MultiServerMCPClient(
        {
            "github": {
                "transport": "http",
                "url": "https://mcp.internal.company.com/github",
                "headers": {"Authorization": "Bearer ghp_..."},
            },
            "slack": {
                "transport": "http",
                "url": "https://mcp.internal.company.com/slack",
                "headers": {"Authorization": "Bearer xoxb-..."},
            },
            "database": {
                "transport": "http",
                "url": "https://mcp.internal.company.com/postgres",
                "headers": {"Authorization": "Bearer db_token_..."},
            },
        }
    )

    # Load all MCP tools
    mcp_tools = await client.get_tools()
    print(f"Loaded {len(mcp_tools)} tools from 3 MCP servers")

    # Create a Deep Agent with MCP tools
    agent = create_deep_agent(
        model="anthropic:claude-sonnet-4-6",
        tools=mcp_tools,
        system_prompt="""You are a DevOps assistant with access to GitHub, Slack, and the database.
Use the appropriate tools to help the user manage their infrastructure.""",
    )

    result = agent.invoke(
        {"messages": [{"role": "user", "content": "Check for open PRs with failing CI, then post a summary to #engineering"}]}
    )
    print(result["messages"][-1].content)

asyncio.run(production_mcp_agent())