"""Example: LangGraph integration with MCP server.
This example demonstrates how to integrate the local MCP server with LangGraph.
Requirements:
pip install langgraph langchain-anthropic mcp
Usage:
python examples/langgraph_integration.py
"""
import asyncio
import json
import sys
from pathlib import Path
from langchain_core.tools import StructuredTool
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def create_mcp_tools():
"""
Create LangGraph tools from MCP server.
Returns:
List of LangGraph StructuredTool objects
"""
# Path to server
server_script_path = str(Path(__file__).parent.parent / "src" / "server.py")
# Server parameters for stdio
server_params = StdioServerParameters(
command=sys.executable, # Python executable
args=[server_script_path],
env=None,
)
# Connect to MCP server
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# Initialize session
await session.initialize()
# Get available tools from MCP
tools_list = await session.list_tools()
print(f"Found {len(tools_list.tools)} MCP tools")
# Convert MCP tools to LangGraph tools
langgraph_tools = []
for mcp_tool in tools_list.tools:
print(f" - {mcp_tool.name}: {mcp_tool.description}")
# Create a closure to capture the tool name and session
def create_tool_func(tool_name):
async def tool_func(**kwargs):
"""Call MCP tool."""
result = await session.call_tool(tool_name, kwargs)
# Parse JSON result
try:
return json.loads(result.content[0].text)
except:
return result.content[0].text
return tool_func
# Convert input schema to LangGraph format
# MCP uses JSON Schema, which LangGraph also accepts
langgraph_tool = StructuredTool(
name=mcp_tool.name,
description=mcp_tool.description,
coroutine=create_tool_func(mcp_tool.name),
args_schema=None, # Let LangGraph infer from function signature
)
langgraph_tools.append(langgraph_tool)
return langgraph_tools, session
async def main():
"""Main function demonstrating LangGraph + MCP integration."""
print("=" * 60)
print("LangGraph + Local MCP Server Integration Example")
print("=" * 60)
print()
# Create MCP tools
print("Connecting to MCP server...")
# Note: In production, you'd want to manage the session lifecycle differently
# This is simplified for demonstration purposes
server_script_path = str(Path(__file__).parent.parent / "src" / "server.py")
server_params = StdioServerParameters(
command=sys.executable,
args=[server_script_path],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
print("Connected! Getting tools...")
tools_list = await session.list_tools()
print(f"\nAvailable tools ({len(tools_list.tools)}):")
for tool in tools_list.tools:
print(f" - {tool.name}")
print("\n" + "=" * 60)
print("Example 1: Calculator Operations")
print("=" * 60)
# Test calculator tools
print("\nCalling calculator_add(5, 3)...")
result = await session.call_tool("calculator_add", {"a": 5, "b": 3})
print(f"Result: {result.content[0].text}")
print("\nCalling calculator_power(2, 10)...")
result = await session.call_tool("calculator_power", {"base": 2, "exponent": 10})
print(f"Result: {result.content[0].text}")
print("\n" + "=" * 60)
print("Example 2: File Operations")
print("=" * 60)
# Test file operations
print("\nCreating test file...")
result = await session.call_tool(
"file_write",
{
"file_path": "test.txt",
"content": "Hello from MCP!\nThis is a test file.\n",
},
)
print(f"Result: {result.content[0].text}")
print("\nReading file...")
result = await session.call_tool("file_read", {"file_path": "test.txt"})
print(f"Result: {result.content[0].text}")
print("\nListing files...")
result = await session.call_tool("file_list", {"directory": "."})
print(f"Result: {result.content[0].text}")
print("\n" + "=" * 60)
print("Example 3: With LangGraph Agent (Optional)")
print("=" * 60)
print("\nTo use with LangGraph agent:")
print("1. Set ANTHROPIC_API_KEY environment variable")
print("2. Uncomment the agent code below")
print("3. Run this script again")
# Uncomment to test with actual LangGraph agent
"""
# Create LLM
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")
# Convert MCP tools to LangGraph format
def create_tool_func(tool_name, session_ref):
async def tool_func(**kwargs):
result = await session_ref.call_tool(tool_name, kwargs)
try:
return json.loads(result.content[0].text)
except:
return result.content[0].text
return tool_func
tools = []
for mcp_tool in tools_list.tools:
tool = StructuredTool(
name=mcp_tool.name,
description=mcp_tool.description,
coroutine=create_tool_func(mcp_tool.name, session),
)
tools.append(tool)
# Create agent
agent = create_react_agent(llm, tools)
# Run agent with a query
query = "Calculate 15 * 7, then create a file called result.txt with the answer"
print(f"\nAgent query: {query}")
async for event in agent.astream({"messages": [("user", query)]}):
print(event)
"""
print("\n" + "=" * 60)
print("Example completed!")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(main())
