### Install Dependencies for Quickstart Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Install necessary libraries and set environment variables for the quickstart example. ```bash pip install langchain-mcp-adapters langgraph "langchain[openai]" export OPENAI_API_KEY= ``` -------------------------------- ### Start Streamable HTTP Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Command to start an example streamable HTTP MCP server. ```bash cd examples/servers/streamable-http-stateless/ uv run mcp-simple-streamablehttp-stateless --port 3000 ``` -------------------------------- ### Install LangChain MCP Adapters Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Install the library using pip. ```bash pip install langchain-mcp-adapters ``` -------------------------------- ### Complete MultiServerMCPClient Configuration Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Provides a full example of setting up the MultiServerMCPClient with multiple server configurations, custom callbacks, and tool interceptors. This setup allows for managing different MCP servers with distinct transport mechanisms and custom logic. ```python from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.callbacks import Callbacks from langchain_mcp_adapters.interceptors import ToolCallInterceptor from datetime import timedelta # Define callbacks async def on_logging(params, context): print(f"[{context.server_name}] {params.level}: {params.message}") async def on_progress(progress, total, message, context): print(f"Progress: {progress}/{total} - {message}") callbacks = Callbacks(on_logging_message=on_logging, on_progress=on_progress) # Define interceptor class TraceInterceptor: async def __call__(self, request, handler): request = request.override( headers={**(request.headers or {}), "X-Trace-ID": "abc123"} ) return await handler(request) # Create client with full configuration client = MultiServerMCPClient( { "math": { "transport": "stdio", "command": "python", "args": ["/path/to/math_server.py"], "env": {"PYTHONUNBUFFERED": "1"}, }, "weather": { "transport": "http", "url": "http://localhost:8000/mcp", "headers": {"Authorization": "Bearer TOKEN"}, "timeout": 10, }, "ai": { "transport": "streamable_http", "url": "http://localhost:3000/mcp", "timeout": timedelta(seconds=60), "sse_read_timeout": timedelta(minutes=5), }, }, callbacks=callbacks, tool_interceptors=[TraceInterceptor()], tool_name_prefix=True, handle_tool_errors=True, ) # Use client tools = await client.get_tools() ``` -------------------------------- ### Complete Callbacks Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/callbacks.md Demonstrates setting up and using multiple callback handlers (logging, progress, elicitation) with the MultiServerMCPClient. This example shows how to define the handlers and pass them to the client configuration. ```python from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext from langchain_mcp_adapters.client import MultiServerMCPClient # Define callback handlers async def on_logging_message(params, context: CallbackContext): """Log messages from the MCP server.""" print(f"[{context.server_name}] {params.level}: {params.message}") async def on_progress(progress, total, message, context: CallbackContext): """Track long-running operations.""" if total: percentage = (progress / total) * 100 print(f"[{context.server_name}] {percentage:.1f}% - {message}") else: print(f"[{context.server_name}] Progress: {progress} - {message}") async def on_elicitation(mcp_context, params, context: CallbackContext): """Handle server requests for user input.""" import getpass if params.isPassword: return getpass.getpass(params.prompt) else: return input(params.prompt) # Create callbacks configuration callbacks = Callbacks( on_logging_message=on_logging_message, on_progress=on_progress, on_elicitation=on_elicitation, ) # Use with client client = MultiServerMCPClient( { "server1": { "transport": "http", "url": "http://localhost:8000/mcp", } }, callbacks=callbacks, ) # Callbacks are invoked automatically during tool operations tools = await client.get_tools() ``` -------------------------------- ### Complete Configuration Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt Demonstrates a complete configuration for the MultiServerMCPClient, including connection, callbacks, and tool interceptors. ```python from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.connections import StreamableHttpConnection from langchain_mcp_adapters.callbacks import Callbacks from langchain_mcp_adapters.interceptors import LoggingInterceptor, RetryInterceptor # Example configuration client = MultiServerMCPClient( connection=StreamableHttpConnection(url="http://localhost:8000"), callbacks=Callbacks(progress_callback=lambda p: print(f"Progress: {p}")), tool_interceptors=[ LoggingInterceptor(), RetryInterceptor(stop_after_attempt=3, wait_fixed=1) ], tool_name_prefix="mcp_", handle_tool_errors=True ) # You can then use the client to interact with MCP services # async def main(): # response = await client.get_tools() # print(response) # import asyncio # asyncio.run(main()) ``` -------------------------------- ### Client Setup with Callbacks Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Configure a MultiServerMCPClient with custom logging and progress callbacks. Ensure your handler functions are defined. ```python from langchain_mcp_adapters.callbacks import Callbacks callbacks = Callbacks( on_logging_message=my_logging_handler, on_progress=my_progress_handler, ) client = MultiServerMCPClient({...}, callbacks=callbacks) ``` -------------------------------- ### Example Progress Callback Implementation Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/callbacks.md An example of how to implement a progress callback function. This function logs the progress percentage and a message, including the server name from the context. ```python async def my_progress_callback(progress, total, message, context: CallbackContext): if total: percent = (progress / total) * 100 print(f"[{context.server_name}] {percent:.1f}%: {message}") else: print(f"[{context.server_name}] Progress {progress}: {message}") callbacks = Callbacks(on_progress=my_progress_callback) ``` -------------------------------- ### Basic Client Setup Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Initialize a MultiServerMCPClient for a single server using stdio transport. This is useful for local script execution. ```python from langchain_mcp_adapters.client import MultiServerMCPClient client = MultiServerMCPClient({ "math": { "transport": "stdio", "command": "python", "args": ["/path/to/math_server.py"], } }) tools = await client.get_tools() ``` -------------------------------- ### Start Weather Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Command to start the weather MCP server using HTTP transport. ```bash python weather_server.py ``` -------------------------------- ### Explicit Session for MultiServerMCPClient Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Demonstrates how to explicitly start a client session for a specific server within MultiServerMCPClient. ```python from langchain_mcp_adapters.tools import load_mcp_tools client = MultiServerMCPClient({...}) async with client.session("math") as session: tools = await load_mcp_tools(session) ``` -------------------------------- ### LangGraph Agent Integration with MCP Tools Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Example of integrating MCP tools into a LangGraph agent. This setup uses a `ToolNode` to handle tool calls within the graph, enabling complex agentic workflows. ```python from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.graph import StateGraph, MessagesState, START from langgraph.prebuilt import ToolNode, tools_condition client = MultiServerMCPClient({...}) tools = await client.get_tools() model = init_chat_model("openai:gpt-4") def call_model(state): return {"messages": [model.bind_tools(tools).invoke(state["messages"])]} graph = StateGraph(MessagesState) graph.add_node("model", call_model) graph.add_node("tools", ToolNode(tools)) graph.add_edge(START, "model") graph.add_conditional_edges("model", tools_condition) graph.add_edge("tools", "model") result = await graph.compile().ainvoke({"messages": [...]}) ``` -------------------------------- ### Run MCP Server with Custom Port Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/examples/servers/streamable-http-stateless/README.md Starts the MCP server on a custom port, for example, port 3000. Use this to avoid port conflicts or to specify a different network interface. ```bash uv run mcp-simple-streamablehttp-stateless --port 3000 ``` -------------------------------- ### Example of Interceptor Chaining Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/interceptors.md Demonstrates how to instantiate a MultiServerMCPClient with a list of tool interceptors. The order in the list dictates the execution flow, from outermost to innermost. ```python client = MultiServerMCPClient( {...}, tool_interceptors=[ LoggingInterceptor(), # Outermost HeaderInjectionInterceptor(...), RetryInterceptor(max_retries=2), CachingInterceptor(), # Innermost ], ) ``` -------------------------------- ### Retrieve Tools from MCP Servers Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/MultiServerMCPClient.md Fetch LangChain `BaseTool` objects from MCP servers. By default, `get_tools()` retrieves tools from all configured servers. You can specify a `server_name` to get tools from a particular server. The example also shows how to integrate these tools with LangGraph for agent creation. ```python # Get all tools from all servers all_tools = await client.get_tools() # Get tools only from the "math" server math_tools = await client.get_tools(server_name="math") # Use tools with LangGraph from langchain.agents import create_agent agent = create_agent("openai:gpt-4.1", all_tools) result = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"}) ``` -------------------------------- ### Environment Variable Expansion Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Demonstrates how environment variables are expanded in configuration. Only braced syntax `${VAR_NAME}` is expanded. Undefined variables are preserved. Values are expanded from the current process environment. ```json { "transport": "stdio", "command": "python", "args": ["/path/to/server.py"], "env": { "API_KEY": "${MY_API_KEY}", # Expanded from os.environ "PASSWORD": "$PASSWORD", # NOT expanded (bare syntax) "MISSING": "${MISSING_VAR}", # Not expanded, stays as is "PYTHONUNBUFFERED": "1" # No expansion needed } } ``` -------------------------------- ### Example Elicitation Callback Implementation Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/callbacks.md An example of an elicitation callback that prompts the user for input, handling password fields securely using getpass. The response is returned, and the SDK wraps it in MCPElicitResult. ```python import getpass async def my_elicitation_callback(mcp_context, params, context: CallbackContext): prompt = params.prompt is_password = params.isPassword if is_password: response = getpass.getpass(prompt) else: response = input(prompt) return response # Returns MCPElicitResult callbacks = Callbacks(on_elicitation=my_elicitation_callback) ``` -------------------------------- ### Logging Interceptor Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt An example of a tool call interceptor that logs requests and responses. This demonstrates the interceptor chaining pattern. ```python from langchain_mcp_adapters.interceptors import ToolCallInterceptor, MCPToolCallRequest, MCPToolCallResult class LoggingInterceptor(ToolCallInterceptor): async def intercept( self, request: MCPToolCallRequest, call_next: ToolCallInterceptor, ) -> MCPToolCallResult: print(f"Logging request: {request.tool_call.name}") result = await call_next(request) print(f"Logging result: {result.tool_call_id}") return result # Usage example would involve configuring this interceptor with MultiServerMCPClient ``` -------------------------------- ### Initialize MultiServerMCPClient with Connections Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/MultiServerMCPClient.md Instantiate the MultiServerMCPClient by providing a dictionary of server connections. Each connection can be configured with different transport types like stdio or http. This example shows setting up connections for a 'math' server using stdio and a 'weather' server using http. ```python from langchain_mcp_adapters.client import MultiServerMCPClient client = MultiServerMCPClient( { "math": { "command": "python", "args": ["/path/to/math_server.py"], "transport": "stdio", }, "weather": { "url": "http://localhost:8000/mcp", "transport": "http", } } ) all_tools = await client.get_tools() ``` -------------------------------- ### Load MCP Tools Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt Loads tools from the MCP SDK. Ensure the MCP SDK is installed and configured. ```python from langchain_mcp_adapters.tools import load_mcp_tools # Load tools from the MCP SDK tools = load_mcp_tools() ``` -------------------------------- ### LoggingMessageCallback Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/callbacks.md Implement a custom callback to handle logging messages from the MCP server. This example shows how to extract log level, message, and logger name from the parameters. ```python async def my_logging_callback(params, context: CallbackContext): level = params.level message = params.message logger = params.logger or "default" print(f"[{context.server_name}/{logger}] {level}: {message}") callbacks = Callbacks(on_logging_message=my_logging_callback) ``` -------------------------------- ### Single Server, Single Session Initialization Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Example of setting up a single MCP server connection using stdio transport and managing a client session. This pattern is useful for dedicated server processes. ```python from mcp import ClientSession from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools params = StdioServerParameters(command="python", args=["/path/to/server.py"]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await load_mcp_tools(session) result = await tools[0].ainvoke({...}) ``` -------------------------------- ### CallbackContext Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/callbacks.md Demonstrates how to access server and tool names within a callback function using CallbackContext. This context is passed to all callback handlers. ```python async def my_callback(params, context: CallbackContext): if context.tool_name: print(f"Tool {context.tool_name} from server {context.server_name}") else: print(f"Event from server {context.server_name}") ``` -------------------------------- ### Routing Interceptor Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt An example of a tool call interceptor that routes tool calls to different backends based on certain criteria. This enables dynamic routing strategies. ```python from langchain_mcp_adapters.interceptors import ToolCallInterceptor, MCPToolCallRequest, MCPToolCallResult class RoutingInterceptor(ToolCallInterceptor): async def intercept( self, request: MCPToolCallRequest, call_next: ToolCallInterceptor, ) -> MCPToolCallResult: # Example: route based on tool name if request.tool_call.name == "specific_tool": # Route to a different client or configuration pass return await call_next(request) # Usage example would involve configuring this interceptor with MultiServerMCPClient ``` -------------------------------- ### Install WebSocket Dependency Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/errors.md If an ImportError occurs due to missing WebSocket support, install the required dependency using pip. ```bash pip install mcp[ws] # or pip install websockets ``` -------------------------------- ### Run MCP Server with Default Port Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/examples/servers/streamable-http-stateless/README.md Starts the MCP server using the default port 3000. This is the basic command to launch the server. ```bash uv run mcp-simple-streamablehttp-stateless ``` -------------------------------- ### Configure Stdio Transport Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Example configuration for the Stdio transport, specifying the command, arguments, and optional environment variables, working directory, and encoding. ```json { "transport": "stdio", "command": "python", "args": ["/path/to/server.py"], "env": {"PYTHONUNBUFFERED": "1"}, "cwd": "/home/user/project", "encoding": "utf-8", } ``` -------------------------------- ### Streamable HTTP Transport Configuration Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Configure the streamable HTTP transport using aliases like 'http'. Useful for standard HTTP endpoints that support streaming. ```python from datetime import timedelta { "transport": "http", "url": "http://localhost:3000/mcp", "timeout": timedelta(seconds=60), "sse_read_timeout": timedelta(minutes=5), "headers": {"X-API-Key": "secret"}, } ``` -------------------------------- ### Integration with LangGraph Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Demonstrates how to integrate the MCP client with LangGraph for building agentic workflows. This example sets up a graph that calls a model and uses tools provided by the MCP client. ```python from langgraph.graph import StateGraph, MessagesState, START from langgraph.prebuilt import ToolNode, tools_condition client = MultiServerMCPClient({...}) tools = await client.get_tools() model = init_chat_model("openai:gpt-4") def call_model(state: MessagesState): response = model.bind_tools(tools).invoke(state["messages"]) return {"messages": [response]} builder = StateGraph(MessagesState) builder.add_node("model", call_model) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "model") builder.add_conditional_edges("model", tools_condition) builder.add_edge("tools", "model") graph = builder.compile() result = await graph.ainvoke({"messages": "..."}) ``` -------------------------------- ### Client Setup with Tool Interceptors Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Integrate custom tool call interceptors to modify requests or transform results before they are processed by the agent. The interceptor must be an async callable. ```python from langchain_mcp_adapters.interceptors import ToolCallInterceptor class MyInterceptor: async def __call__(self, request, handler): # Modify request request = request.override(args={...}) # Call handler result = await handler(request) # Transform result return result client = MultiServerMCPClient( {...}, tool_interceptors=[MyInterceptor()] ) ``` -------------------------------- ### Example of Valid MCP Prompt Messages for LangChain Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/prompts.md Demonstrates the structure of valid MCP prompt messages that can be converted to LangChain messages. Requires role 'user' or 'assistant' and content type 'text'. ```python from mcp.types import PromptMessage, TextContent from langchain_mcp_adapters.prompts import load_mcp_prompt # Valid prompt messages (role: user or assistant, content: text) async def valid_example(session): messages = await load_mcp_prompt(session, "chat") # Returns a list of HumanMessage/AIMessage objects # The function handles conversion automatically: # MCP PromptMessage(role="user", content=TextContent(text="...")) # → HumanMessage(content="...") # MCP PromptMessage(role="assistant", content=TextContent(text="...")) # → AIMessage(content="...") ``` -------------------------------- ### Run MCP Server with Custom Logging Level Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/examples/servers/streamable-http-stateless/README.md Starts the MCP server with a custom logging level, such as DEBUG. This is useful for detailed troubleshooting and monitoring. ```bash uv run mcp-simple-streamablehttp-stateless --log-level DEBUG ``` -------------------------------- ### Integrate with LangGraph for Agentic Workflows Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/INDEX.md Set up a LangGraph agent that utilizes MCP tools for its actions. This involves initializing the client, getting tools, and defining graph nodes for model calls and tool execution. ```python from langgraph.graph import StateGraph, MessagesState, START from langgraph.prebuilt import ToolNode, tools_condition client = MultiServerMCPClient({...}) tools = await client.get_tools() def call_model(state): model = init_chat_model("openai:gpt-4") return {"messages": [model.bind_tools(tools).invoke(state["messages"])]} graph = StateGraph(MessagesState) graph.add_node("model", call_model) graph.add_node("tools", ToolNode(tools)) # ... add edges ... result = await graph.compile().ainvoke({"messages": [...]}) ``` -------------------------------- ### Load All MCP Resources Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/resources.md Loads all static resources from an MCP client session. Use this to get all available resources when specific URIs are not needed. Dynamic resources are not included. ```python from mcp import ClientSession from mcp.client.stdio import stdio_client from langchain_mcp_adapters.resources import load_mcp_resources # Create and initialize session server_params = StdioServerParameters( command="python", args=["/path/to/resource_server.py"], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # Load all resources all_resources = await load_mcp_resources(session) # Load specific resources by URI specific = await load_mcp_resources( session, uris=[ "file:///documents/readme.txt", "file:///data/image.png" ] ) # Load a single resource single = await load_mcp_resources( session, uris="file:///config/settings.json" ) # Access blob data for blob in all_resources: print(f"Resource: {blob.metadata['uri']}") print(f"Size: {len(blob.as_bytes())} bytes") print(f"MIME type: {blob.mime_type}") ``` -------------------------------- ### Load All Tools from All Servers Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/INDEX.md Instantiate the client with multiple server configurations and load all available tools. ```python client = MultiServerMCPClient({"server1": {...}, "server2": {...}}) all_tools = await client.get_tools() ``` -------------------------------- ### Retry Interceptor Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt An example of a tool call interceptor that retries failed tool calls. This is useful for handling transient errors. ```python from langchain_mcp_adapters.interceptors import ToolCallInterceptor, MCPToolCallRequest, MCPToolCallResult from tenacity import retry, stop_after_attempt, wait_fixed class RetryInterceptor(ToolCallInterceptor): @retry(stop=stop_after_attempt(3), wait=wait_fixed(1)) async def intercept( self, request: MCPToolCallRequest, call_next: ToolCallInterceptor, ) -> MCPToolCallResult: return await call_next(request) # Usage example would involve configuring this interceptor with MultiServerMCPClient ``` -------------------------------- ### Caching Interceptor Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt An example of a tool call interceptor that caches tool call results. This can improve performance by avoiding redundant computations. ```python from langchain_mcp_adapters.interceptors import ToolCallInterceptor, MCPToolCallRequest, MCPToolCallResult class CachingInterceptor(ToolCallInterceptor): def __init__(self): self.cache = {} async def intercept( self, request: MCPToolCallRequest, call_next: ToolCallInterceptor, ) -> MCPToolCallResult: cache_key = (request.tool_call.name, request.tool_call.args) if cache_key in self.cache: return self.cache[cache_key] result = await call_next(request) self.cache[cache_key] = result return result # Usage example would involve configuring this interceptor with MultiServerMCPClient ``` -------------------------------- ### Headers Interceptor Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt An example of a tool call interceptor that adds custom headers to the tool call request. This is useful for passing metadata or authentication tokens. ```python from langchain_mcp_adapters.interceptors import ToolCallInterceptor, MCPToolCallRequest, MCPToolCallResult class HeadersInterceptor(ToolCallInterceptor): def __init__(self, headers: dict): self.headers = headers async def intercept( self, request: MCPToolCallRequest, call_next: ToolCallInterceptor, ) -> MCPToolCallResult: request.headers.update(self.headers) return await call_next(request) # Usage example would involve configuring this interceptor with MultiServerMCPClient ``` -------------------------------- ### MultiServerMCPClient Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt Documentation for the main client class, including its constructor, session management, and methods for retrieving tools, prompts, and resources. ```APIDOC ## Class: MultiServerMCPClient ### Description This is the main client class for interacting with MCP services. It provides methods to manage sessions and retrieve various resources like tools, prompts, and other assets. ### Constructor Initializes the client with various configuration options. See `configuration.md` for detailed parameter documentation. ### Methods #### `session()` ##### Description Provides a context manager for managing client sessions. #### `get_tools()` ##### Description Retrieves available tools from the MCP service. #### `get_prompt()` ##### Description Retrieves prompts from the MCP service. #### `get_resources()` ##### Description Retrieves resources from the MCP service. ``` -------------------------------- ### Initialize MCP Client and Load Tools Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Initializes a streamable HTTP client to connect to an MCP server, loads MCP tools, and creates an agent. Use this when directly interacting with a streamable HTTP MCP server. ```python from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client from langchain.agents import create_agent from langchain_mcp_adapters.tools import load_mcp_tools async with streamablehttp_client("http://localhost:3000/mcp") as (read, write, _): async with ClientSession(read, write) as session: # Initialize the connection await session.initialize() # Get tools tools = await load_mcp_tools(session) agent = create_agent("openai:gpt-4.1", tools) math_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"}) ``` -------------------------------- ### Load Tools from a Single Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/INDEX.md Instantiate the MultiServerMCPClient and load tools from a specific server using its configuration. ```python from langchain_mcp_adapters.client import MultiServerMCPClient client = MultiServerMCPClient({ "math": { "transport": "stdio", "command": "python", "args": ["/path/to/math_server.py"], } }) tools = await client.get_tools() ``` -------------------------------- ### Selective Tool Interceptor Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/interceptors.md Example of a SelectiveToolInterceptor that returns a ToolMessage early for specific deprecated tools. For other tools, it proceeds with normal execution via the handler. This demonstrates conditional short-circuiting. ```python from mcp.types import CallToolResult from langchain_core.messages import ToolMessage class SelectiveToolInterceptor: async def __call__(self, request: MCPToolCallRequest, handler): # For specific tools, return early with ToolMessage if request.name == "deprecated_tool": return ToolMessage( content="This tool is deprecated. Use new_tool instead.", tool_call_id="error", ) # For others, execute normally return await handler(request) ``` -------------------------------- ### Load Tools from a Specific Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/INDEX.md Instantiate the client and load tools from a particular server by specifying its name. ```python client = MultiServerMCPClient({...}) server_tools = await client.get_tools(server_name="math") ``` -------------------------------- ### Load Tools from Multiple Servers Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/INDEX.md Configure the client to connect to multiple servers simultaneously and retrieve tools from all or specific servers. ```python client = MultiServerMCPClient({ "math": {"transport": "stdio", "command": "python", "args": [...]}, "weather": {"transport": "http", "url": "http://localhost:8000/mcp"}, }) all_tools = await client.get_tools() math_tools = await client.get_tools(server_name="math") ``` -------------------------------- ### Load Static and Dynamic MCP Resources Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/resources.md Illustrates how to load resources from an MCP session. Use `uris=None` to load only static resources, or provide specific URIs to load dynamic resources, which may require parameters. ```python # Load only static resources static = await load_mcp_resources(session, uris=None) # Load specific dynamic resource dynamic = await load_mcp_resources( session, uris="file:///dynamic/resource?param=value" ) ``` -------------------------------- ### Get MCP Resource Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/MANIFEST.txt Retrieves a specific resource from the MCP SDK. Useful for accessing individual data assets. ```python from langchain_mcp_adapters.resources import get_mcp_resource # Get a single resource by its identifier resource = get_mcp_resource("resource_id") ``` -------------------------------- ### Multiple Servers with Auto-Session Creation Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Demonstrates using `MultiServerMCPClient` to connect to multiple servers with different transports (stdio and http). Sessions are automatically created per tool call when not explicitly managed. ```python from langchain_mcp_adapters.client import MultiServerMCPClient client = MultiServerMCPClient({ "math": {"transport": "stdio", "command": "python", "args": [...]}, "weather": {"transport": "http", "url": "http://localhost:8000/mcp"}, }) tools = await client.get_tools() # Auto-creates sessions result = await tools[0].ainvoke({...}) # Each call creates new session ``` -------------------------------- ### Integrate MCP Resources with LangChain Document Loaders Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/resources.md Demonstrates how to load MCP resources using `MultiServerMCPClient`, convert them into LangChain `Document` objects, and then use these documents with LangChain's retrieval chains, such as FAISS for vector storage. ```python from langchain_core.documents import Document from langchain_mcp_adapters.client import MultiServerMCPClient client = MultiServerMCPClient({ "docs": { "transport": "http", "url": "http://localhost:8000/mcp", } }) # Load resources blobs = await client.get_resources(server_name="docs") # Convert to LangChain Documents documents = [ Document( page_content=blob.as_string(), metadata={"uri": blob.metadata["uri"], "mime_type": blob.mime_type} ) for blob in blobs ] # Use with retrieval chains from langchain.vectorstores import FAISS from langchain.embeddings.openai import OpenAIEmbeddings embeddings = OpenAIEmbeddings() vectorstore = FAISS.from_documents(documents, embeddings) ``` -------------------------------- ### MCP Client with Math Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Connects to an MCP math server via stdio, loads tools, and uses them with a LangChain agent. ```python # Create server parameters for stdio connection from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain.agents import create_agent server_params = StdioServerParameters( command="python", # Make sure to update to the full absolute path to your math_server.py file args=["/path/to/math_server.py"], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize the connection await session.initialize() # Get tools tools = await load_mcp_tools(session) # Create and run the agent agent = create_agent("openai:gpt-4.1", tools) agent_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"}) ``` -------------------------------- ### Initialize MultiServerMCPClient and Load Tools Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Initializes a MultiServerMCPClient with HTTP transport for a math tool and creates an agent. This is useful for managing multiple MCP servers or services. ```python # Use server from examples/servers/streamable-http-stateless/ from langchain_mcp_adapters.client import MultiServerMCPClient from langchain.agents import create_agent client = MultiServerMCPClient( { "math": { "transport": "http", "url": "http://localhost:3000/mcp" }, } ) tools = await client.get_tools() agent = create_agent("openai:gpt-4.1", tools) math_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"}) ``` -------------------------------- ### Custom Tool Interceptor Example Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/INDEX.md Implement a custom interceptor class with pre- and post-execution logic for tool calls, including request modification and error handling. ```python class CustomInterceptor: async def __call__(self, request, handler): # Pre-execution logic request = request.override(args={...}) try: result = await handler(request) except Exception: # Error handling raise # Post-execution logic return result client = MultiServerMCPClient({...}, tool_interceptors=[CustomInterceptor()]) ``` -------------------------------- ### Custom Tool Interceptor for Request Modification Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Illustrates how to add custom interceptors to `MultiServerMCPClient` to modify requests before they are sent. This example adds an `Authorization` header to all requests. ```python class HeaderInterceptor: def __init__(self, headers): self.headers = headers async def __call__(self, request, handler): request = request.override( headers={**(request.headers or {}), **self.headers} ) return await handler(request) client = MultiServerMCPClient( {...}, tool_interceptors=[ HeaderInterceptor({"Authorization": "Bearer TOKEN"}) ] ) ``` -------------------------------- ### Configure StdioConnection with Environment Variable Expansion Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/sessions.md Configure a StdioConnection where environment variables in the 'env' dictionary values can be expanded using the ${VAR} syntax. Only braced syntax is supported. ```python connection: StdioConnection = { "transport": "stdio", "command": "python", "args": ["/path/to/server.py"], "env": { "PYTHONUNBUFFERED": "1", "API_KEY": "${MY_API_KEY}", # Expanded "PASSWORD": "$PASSWORD", # NOT expanded (bare syntax) }, } ``` -------------------------------- ### Detect Unexpanded Environment Variable Reference Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/errors.md This example demonstrates how a warning is logged when an environment variable reference in `StdioConnection.env` is undefined. The unexpanded variable is passed as-is to the subprocess. ```python from langchain_mcp_adapters.client import MultiServerMCPClient import logging logging.basicConfig(level=logging.WARNING) client = MultiServerMCPClient({ "server": { "transport": "stdio", "command": "python", "args": ["server.py"], "env": {"API_KEY": "${MISSING_VAR}"}, # Will warn } }) # Warning: env['API_KEY'] contains unexpanded variable reference: '${MISSING_VAR}' ``` -------------------------------- ### Create an MCP Math Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Defines an MCP server with 'add' and 'multiply' tools using FastMCP and runs it over stdio transport. ```python # math_server.py from mcp.server.fastmcp import FastMCP 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__": mcp.run(transport="stdio") ``` -------------------------------- ### LangGraph Workflow with MCP Prompts Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/prompts.md This snippet illustrates setting up a LangGraph workflow that loads system context from an MCP server before invoking a chat model. Ensure the MultiServerMCPClient is configured with the correct transport and command for your prompt server. ```python from langchain_mcp_adapters.client import MultiServerMCPClient from langchain.chat_models import init_chat_model from langgraph.graph import StateGraph, MessagesState, START client = MultiServerMCPClient({ "assistant": { "transport": "stdio", "command": "python", "args": ["/path/to/prompt_server.py"], } }) async def load_system_context(state: MessagesState): # Load system prompt from MCP server async with client.session("assistant") as session: system_messages = await load_mcp_prompt( session, "system_context", arguments={"role": "helpful_assistant"} ) # Prepend system messages to state return {"messages": system_messages + state["messages"]} async def model_call(state: MessagesState): model = init_chat_model("openai:gpt-4") response = model.invoke(state["messages"]) return {"messages": [response]} builder = StateGraph(MessagesState) builder.add_node("load_context", load_system_context) builder.add_node("model", model_call) builder.add_edge(START, "load_context") builder.add_edge("load_context", "model") graph = builder.compile() ``` -------------------------------- ### Request Transformation Interceptor Implementation and Usage Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/interceptors.md Implement a NormalizeArgsInterceptor to transform tool arguments before execution. This example normalizes string arguments to lowercase. Initialize the client with this interceptor to preprocess arguments. ```python class NormalizeArgsInterceptor: """Normalize tool arguments before execution.""" async def __call__(self, request: MCPToolCallRequest, handler): # Transform arguments (e.g., convert to expected types) normalized_args = { k: (v.lower() if isinstance(v, str) else v) for k, v in request.args.items() } request = request.override(args=normalized_args) return await handler(request) client = MultiServerMCPClient( {...}, tool_interceptors=[NormalizeArgsInterceptor()], ) ``` -------------------------------- ### Interceptor Example: Modifying Request Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/interceptors.md Demonstrates how an interceptor can access request context, modify modifiable fields like arguments and headers using `override()`, and then call the next handler in the chain. ```python from langchain_mcp_adapters.interceptors import MCPToolCallRequest # Inside an interceptor async def my_interceptor(request: MCPToolCallRequest, handler): # Access context print(f"Tool: {request.name} from {request.server_name}") # Modify modifiable fields request = request.override( args={**request.args, "timeout": 30}, headers={**request.headers or {}, "X-Trace-ID": "abc123"} ) # Call handler with modified request return await handler(request) ``` -------------------------------- ### Basic Logging Interceptor Usage Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/interceptors.md Demonstrates the basic usage of a LoggingInterceptor with the MultiServerMCPClient. Ensure the interceptor is included in the tool_interceptors list during client initialization. ```python client = MultiServerMCPClient( {...}, tool_interceptors=[LoggingInterceptor()], ) ``` -------------------------------- ### Run MCP Server with JSON Responses Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/examples/servers/streamable-http-stateless/README.md Starts the MCP server with JSON responses enabled instead of Server-Sent Events (SSE) streams. This option changes the response format for client compatibility. ```bash uv run mcp-simple-streamablehttp-stateless --json-response ``` -------------------------------- ### Configure MultiServerMCPClient with Connections Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Define server connections using a dictionary mapping server names to connection configurations. Supports StdioConnection, SSEConnection, StreamableHttpConnection, or WebsocketConnection. ```python connections = { "math": { "transport": "stdio", "command": "python", "args": ["/path/to/math_server.py"], }, "weather": { "transport": "http", "url": "http://localhost:8000/mcp", } } client = MultiServerMCPClient(connections) ``` -------------------------------- ### Concurrent Tool Loading from Multiple Servers Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Shows how to load tools concurrently from multiple MCP servers using `MultiServerMCPClient`. This leverages `asyncio.gather` internally for parallel loading. ```python # Concurrent loading from multiple servers client = MultiServerMCPClient({ "server1": {...}, "server2": {...}, "server3": {...}, }) # Uses asyncio.gather() internally all_tools = await client.get_tools() # Loads all servers in parallel ``` -------------------------------- ### Get Prompt from MCP Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/MultiServerMCPClient.md Retrieves a prompt from a specified MCP server and converts it into LangChain message objects. Use this method when you need to fetch predefined prompts with optional arguments for dynamic content. ```python async def get_prompt( self, server_name: str, prompt_name: str, *, arguments: dict[str, Any] | None = None, ) -> list[HumanMessage | AIMessage] ``` ```python messages = await client.get_prompt( server_name="assistant", prompt_name="analysis", arguments={"topic": "Python"} ) # messages is now a list of HumanMessage/AIMessage objects ``` -------------------------------- ### MCPToolCallRequest.override Method Examples Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/interceptors.md Illustrates various ways to use the `override` method to create new `MCPToolCallRequest` instances with modified fields like tool name, arguments, or headers. The original request remains immutable. ```python # Modify tool arguments new_request = request.override(args={"value": 10, "timeout": 30}) ``` ```python # Change tool name new_request = request.override(name="different_tool") ``` ```python # Update headers new_request = request.override( headers={**(request.headers or {}), "Authorization": "Bearer TOKEN"} ) ``` ```python # Multiple overrides new_request = request.override( name="new_tool", args={"param": "value"}, headers={"X-Custom": "header"}, ) ``` -------------------------------- ### Create an MCP Weather Server Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/README.md Defines an MCP server with a 'get_weather' tool using FastMCP and runs it over HTTP transport. ```python # weather_server.py from typing import List from mcp.server.fastmcp import FastMCP mcp = FastMCP("Weather") @mcp.tool() async def get_weather(location: str) -> str: """Get weather for location.""" return "It's always sunny in New York" if __name__ == "__main__": mcp.run(transport="http") ``` -------------------------------- ### load_mcp_tools() Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Loads MCP tools, optionally with custom session, callbacks, interceptors, and server configuration. ```APIDOC ## load_mcp_tools() ### Description Loads MCP tools, optionally with custom session, callbacks, interceptors, and server configuration. ### Parameters - **session** (`ClientSession | None`) - MCP session (required if connection not provided). - **connection** (`Connection | None`) - Connection config (required if session not provided). Defaults to `None`. - **callbacks** (`Callbacks | None`) - Callbacks for notifications. Defaults to `None`. - **tool_interceptors** (`list[ToolCallInterceptor] | None`) - Tool interceptors. Defaults to `None`. - **server_name** (`str | None`) - Server identifier. Defaults to `None`. - **tool_name_prefix** (`bool`) - Prefix tool names with server name. Defaults to `False`. - **handle_tool_errors** (`bool`) - Handle errors gracefully. Defaults to `True`. ### Returns `list[BaseTool]` - A list of loaded MCP tools. ``` -------------------------------- ### Configure Callbacks for MCP Events Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Set up optional callbacks to handle MCP server notifications and events, such as logging and progress updates. Ensure the Callbacks class is imported. ```python from langchain_mcp_adapters.callbacks import Callbacks async def on_logging(params, context): print(f"Log: {params.message}") callbacks = Callbacks(on_logging_message=on_logging) client = MultiServerMCPClient({...}, callbacks=callbacks) ``` -------------------------------- ### session Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/MultiServerMCPClient.md Creates an asynchronous context manager for establishing and managing a session with a specific MCP server. This allows for controlled interaction with a server, including automatic or manual session initialization. ```APIDOC ## session ### Description Creates an asynchronous context manager for establishing and managing a session with a specific MCP server. This allows for controlled interaction with a server, including automatic or manual session initialization. ### Method `async def session(self, server_name: str, *, auto_initialize: bool = True) -> AsyncIterator[ClientSession]` ### Parameters #### Server Name - **server_name** (str) - Required - Name identifying the server connection in the connections dictionary. #### Auto Initialize - **auto_initialize** (bool) - Optional - True - Whether to automatically initialize the session. If False, the session must be manually initialized with `await session.initialize()`. ### Returns An async context manager yielding an initialized `ClientSession` from the MCP SDK. ### Raises - `ValueError`: If the server name is not found in the connections dictionary. ### Example ```python from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.tools import load_mcp_tools client = MultiServerMCPClient({...}) async with client.session("math") as session: tools = await load_mcp_tools(session) ``` ``` -------------------------------- ### Inefficient vs. Efficient Session Reuse Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/OVERVIEW.md Demonstrates the difference between creating new sessions for each tool call (inefficient) and reusing a session within an async context (efficient). Session reuse can significantly improve performance. ```python # Inefficient: Creates new session for each tool call tools = await client.get_tools() result1 = await tools[0].ainvoke({...}) # New session result2 = await tools[0].ainvoke({...}) # New session # Efficient: Reuse session async with client.session("server_name") as session: await session.initialize() tools = await load_mcp_tools(session) result1 = await tools[0].ainvoke({...}) # Reuses session result2 = await tools[0].ainvoke({...}) # Reuses session ``` -------------------------------- ### Enable Tool Name Prefixing Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/configuration.md Set `tool_name_prefix=True` to prefix tool names with the server name, preventing naming conflicts when multiple servers have tools with the same name. ```python client = MultiServerMCPClient( { "weather": {...}, "climate": {...}, }, tool_name_prefix=True ) # Tools will be named "weather_search", "climate_search", etc. ``` -------------------------------- ### Manage MCP Server Session with Context Manager Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/api-reference/MultiServerMCPClient.md Use the `session` context manager to establish and manage a connection to a specific MCP server. This is useful for performing multiple operations on the same server, ensuring proper session initialization and cleanup. The `auto_initialize` parameter controls whether the session is automatically prepared. ```python from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.tools import load_mcp_tools client = MultiServerMCPClient({...}) async with client.session("math") as session: tools = await load_mcp_tools(session) ``` -------------------------------- ### Handling Missing Transport Parameters in create_session Source: https://github.com/langchain-ai/langchain-mcp-adapters/blob/main/_autodocs/errors.md Demonstrates how to handle ValueErrors that arise from missing required parameters (like 'transport', 'url', 'command', or 'args') when configuring different connection transports for create_session. ```python from langchain_mcp_adapters.sessions import create_session # Missing transport try: async with create_session({"url": "http://localhost:8000"}) as session: pass except ValueError as e: # "Configuration error: Missing 'transport' key..." pass # Stdio missing command try: async with create_session({ "transport": "stdio", "args": ["script.py"], }) as session: pass except ValueError as e: # "'command' parameter is required..." pass ```