Skip to content

datarobot_genai.langgraph.mcp

mcp

mcp_tools_context async 

mcp_tools_context(mcp_config: MCPConfig) -> AsyncGenerator[list[BaseTool], None]

Yield a list of LangChain BaseTool instances loaded via MCP.

If no configuration or loading fails, yields an empty list without raising.

Parameters

authorization_context : dict[str, Any] | None Authorization context to use for MCP connections forwarded_headers : dict[str, str] | None Forwarded headers, e.g. x-datarobot-api-key to use for MCP authentication

Source code in datarobot_genai/langgraph/mcp.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@asynccontextmanager
async def mcp_tools_context(
    mcp_config: MCPConfig,
) -> AsyncGenerator[list[BaseTool], None]:
    """Yield a list of LangChain BaseTool instances loaded via MCP.

    If no configuration or loading fails, yields an empty list without raising.

    Parameters
    ----------
    authorization_context : dict[str, Any] | None
        Authorization context to use for MCP connections
    forwarded_headers : dict[str, str] | None
        Forwarded headers, e.g. x-datarobot-api-key to use for MCP authentication
    """
    server_config = mcp_config.server_config

    if not server_config:
        logger.info("No MCP server configured, using empty tools list")
        yield []
        return

    # Prevent mutation of the original server_config
    server_config = copy.deepcopy(server_config)

    url = server_config["url"]
    logger.info("Connecting to MCP server: %s", url)

    # Pop transport from server_config to avoid passing it twice
    # Use .pop() with default to never error
    transport = server_config.pop("transport", "streamable-http")

    if transport in ["streamable-http", "streamable_http"]:
        connection = StreamableHttpConnection(transport="streamable_http", **server_config)
    elif transport == "sse":
        connection = SSEConnection(transport="sse", **server_config)
    else:
        raise RuntimeError("Unsupported MCP transport specified.")

    # Graceful fallback: if we can't connect to the MCP server, yield empty tools
    # instead of crashing.
    #
    # The `connected` flag distinguishes two cases for the except clause:
    #   - Exception raised during setup (before yield): log a warning and yield [].
    #   - Exception thrown back in from the consumer (after yield, via athrow()): re-raise.
    # Without this guard, a consumer exception of a caught type would hit `yield []` as a
    # second yield, causing `RuntimeError: generator didn't stop after athrow()`.
    connected = False
    try:
        async with create_session(connection=connection) as session:
            raw_tools = await load_mcp_tools(session=session)
            tools = [_wrap_mcp_tool_for_langgraph(t) for t in raw_tools]
            logger.info("Successfully loaded %d MCP tools", len(tools))
            connected = True
            yield tools
    except (ConnectionError, OSError, TimeoutError, ExceptionGroup) as exc:
        if connected:
            raise
        logger.warning(
            "Failed to connect to MCP server at %s: %s. Continuing without MCP tools.",
            url,
            exc,
        )
        yield []