Source code for langchain_mcp_tools_docs

# Documentation-only version with simplified code
import logging
from typing import Any, Awaitable, Callable, Dict, List, Optional, TextIO, Tuple, TypedDict, Union, NotRequired

# Import for forward reference
try:
    from langchain_core.tools import BaseTool
except ImportError:
    # For documentation only - create a placeholder if import fails
    class BaseTool:
        """Placeholder for LangChain BaseTool."""
        pass

# Type definitions for public API


class McpServerCommandBasedConfig(TypedDict):
    """Configuration for an MCP server launched via command line.

    This configuration is used for local MCP servers that are started as child
    processes using the stdio client. It defines the command to run, optional
    arguments, environment variables, working directory, and error logging
    options.

    Attributes:
        command: The executable command to run (e.g., "npx", "uvx", "python").
        args: Optional list of command-line arguments to pass to the command.
        env: Optional dictionary of environment variables to set for the
            process.
        cwd: Optional working directory where the command will be executed.
        errlog: Optional file-like object for redirecting the server's stderr
            output.

    Example::

        {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "."],
            "env": {"NODE_ENV": "production"},
            "cwd": "/path/to/working/directory",
            "errlog": open("server.log", "w")
        }
    """
    command: str
    args: NotRequired[List[str] | None]
    env: NotRequired[Dict[str, str] | None]
    cwd: NotRequired[str | None]
    errlog: NotRequired[TextIO | None]





class McpServerUrlBasedConfig(TypedDict):
    """Configuration for a remote MCP server accessed via URL.

    This configuration is used for remote MCP servers that are accessed via
    HTTP/HTTPS (Server-Sent Events) or WebSocket connections. It defines the
    URL to connect to and optional HTTP headers for authentication.

    Attributes:
        url: The URL of the remote MCP server. For SSE servers,
            use http:// or https:// prefix. For WebSocket servers, 
            use ws:// or wss:// prefix.
        headers: Optional dictionary of HTTP headers to include in the request,
            typically used for authentication (e.g., bearer tokens).

    Example for SSE server::

        {
            "url": "https://example.com/mcp/sse",
            "headers": {"Authorization": "Bearer token123"}
        }

    Example for WebSocket server::

        {
            "url": "wss://example.com/mcp/ws"
        }
    """
    url: str
    headers: NotRequired[Dict[str, str] | None]



# Type for a single MCP server configuration, which can be either command-based or URL-based.
SingleMcpServerConfig = Union[McpServerCommandBasedConfig, McpServerUrlBasedConfig]
"""Configuration for a single MCP server, either command-based or URL-based.

This type represents the configuration for a single MCP server, which can be either:

1. A local server launched via command line (McpServerCommandBasedConfig)
2. A remote server accessed via URL (McpServerUrlBasedConfig)

The type is determined by the presence of either the "command" key (for command-based)
or the "url" key (for URL-based).
"""

# Configuration dictionary for multiple MCP servers
McpServersConfig = Dict[str, SingleMcpServerConfig]
"""Configuration dictionary for multiple MCP servers.

A dictionary mapping server names (as strings) to their respective configurations.
Each server name acts as a logical identifier used for logging and debugging.
The configuration for each server can be either command-based or URL-based.

Example::

    {
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "."]
        },
        "fetch": {
            "command": "uvx",
            "args": ["mcp-server-fetch"]
        },
        "remote-server": {
            "url": "https://example.com/mcp/sse",
            "headers": {"Authorization": "Bearer token123"}
        }
    }
"""


# Type hint for cleanup function
McpServerCleanupFn = Callable[[], Awaitable[None]]
"""Type for the async cleanup function returned by convert_mcp_to_langchain_tools.

This represents an asynchronous function that takes no arguments and returns
nothing. It's used to properly shut down all MCP server connections and clean
up resources when the tools are no longer needed.

Example usage::

    tools, cleanup = await convert_mcp_to_langchain_tools(server_configs)
    # Use tools...
    await cleanup()  # Clean up resources when done
"""




async def convert_mcp_to_langchain_tools(
    server_configs: McpServersConfig,
    logger: Optional[logging.Logger] = None
) -> Tuple[List[BaseTool], McpServerCleanupFn]:
    """Initialize multiple MCP servers and convert their tools to
    LangChain format.

    This async function manages parallel initialization of multiple MCP
    servers, converts their tools to LangChain format, and provides a cleanup
    mechanism. It orchestrates the full lifecycle of multiple servers.

    Args:
        server_configs: Dictionary mapping server names to their
            configurations, where each configuration contains command, args,
            and env settings
        logger: Logger instance to use for logging events and errors.
            If None, uses module logger with fallback to a pre-configured
            logger when no root handlers exist.

    Returns:
        A tuple containing:

        * List of converted LangChain tools from all servers
        * Async cleanup function to properly shutdown all server connections

    Example::

        server_configs = {
            "fetch": {
                "command": "uvx", "args": ["mcp-server-fetch"]
            },
            "weather": {
                "command": "npx", "args": ["-y","@h1deya/mcp-server-weather"]
            }
        }
        
        tools, cleanup = await convert_mcp_to_langchain_tools(server_configs)
        
        # Use tools...
        
        await cleanup()
    """
    # This is just a documentation stub
    pass