lg:mcp

lg:mcp - Model Context Protocol Integration

Integrate LangGraph agents with MCP to expose agents as servers or consume external MCP tools.

Workflow Steps

Expose Agent as MCP Server

1. Validate Project

   • Check project structure
   • Verify agent is functional
   • Validate configuration

2. Generate MCP Server Code

   • Create server wrapper
   • Define tool schemas
   • Setup transport layer
   • Configure authentication (if enabled)

3. Generate Tool Definitions

   • Extract agent capabilities
   • Create MCP tool schemas
   • Document parameters
   • Define return types

4. Setup Transport

   • Configure stdio/SSE/WebSocket
   • Setup request handling
   • Configure streaming (if enabled)
   • Add error handling

5. Generate Claude Config

   • Create claude_desktop_config.json
   • Add server configuration
   • Document setup instructions

6. Create Tests

   • Add MCP server tests
   • Test tool invocation
   • Test streaming (if enabled)
   • Test error handling

7. Update Documentation

   • Add MCP server docs
   • Document available tools
   • Add integration examples

Consume MCP Tools

1. Connect to MCP Server

   • Load server configuration
   • Establish connection
   • Validate server availability

2. Discover Tools

   • Query available tools
   • Parse tool schemas
   • Validate compatibility

3. Generate Tool Wrappers

   • Create LangChain tool wrappers
   • Map MCP schemas to LangChain
   • Add type validation

4. Integrate with Agent

   • Add tools to agent
   • Update tool list
   • Configure tool binding

5. Update Tests

   • Add tool usage tests
   • Test MCP connection
   • Test tool execution

6. Update Documentation

   • Document available MCP tools
   • Add usage examples

MCP Server Patterns

Stdio Transport (Default)

Best for Claude Desktop integration.

# mcp_server.py
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

from src.graph import app

# Create MCP server
server = Server("langgraph-agent")

@server.list_tools()
async def list_tools() -> list[Tool]:
    """List available tools."""
    return [
        Tool(
            name="run_agent",
            description="Run the LangGraph agent",
            inputSchema={
                "type": "object",
                "properties": {
                    "message": {
                        "type": "string",
                        "description": "Input message"
                    },
                    "thread_id": {
                        "type": "string",
                        "description": "Thread ID for conversation"
                    }
                },
                "required": ["message"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Execute tool."""
    if name == "run_agent":
        message = arguments["message"]
        thread_id = arguments.get("thread_id", "default")

        config = {"configurable": {"thread_id": thread_id}}
        result = app.invoke({"messages": [message]}, config)

        return [
            TextContent(
                type="text",
                text=result["messages"][-1].content
            )
        ]

    raise ValueError(f"Unknown tool: {name}")

async def main():
    """Run MCP server."""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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