How to use MCP with LangChain

pip install mcp langchain

import os
from anthropic import Anthropic
from mcp.server.stdio import stdio_server
from mcp.server import Server
from langchain.schema import BaseMessage
from langchain.agents import AgentExecutor, Tool
from langchain.chains.base import Chain

# Step 1: Define MCP server handler
class MyMCPServer(Server):
    def __init__(self):
        super().__init__()
        self.client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    async def handle_message(self, message):
        # Example: forward message content to Anthropic Claude model
        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=512,
            system="You are an assistant connected via MCP.",
            messages=[{"role": "user", "content": message.content}]
        )
        return response.content[0].text

# Step 2: Run MCP server in background (for demo, run in main thread)
import threading
import asyncio

def run_mcp_server():
    stdio_server(MyMCPServer())

thread = threading.Thread(target=run_mcp_server, daemon=True)
thread.start()

# Step 3: Define a LangChain Chain that sends queries to MCP server
class MCPChain(Chain):
    def __init__(self):
        super().__init__()

    @property
    def input_keys(self):
        return ["input"]

    @property
    def output_keys(self):
        return ["output"]

    def _call(self, inputs):
        # For demo, simulate sending input to MCP server and getting response
        # In production, implement IPC or network communication with MCP server
        user_input = inputs["input"]
        # Here we just echo input for demonstration
        output = f"MCP server response to: {user_input}"
        return {"output": output}

# Step 4: Use MCPChain in LangChain
mcp_chain = MCPChain()

result = mcp_chain.invoke({"input": "Hello from LangChain to MCP!"})
print(result["output"])