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
| @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:
# Load tools WITHOUT holding a persistent MCP session open across the stream.
# ``session=None`` + ``connection`` makes ``load_mcp_tools`` open a short-lived
# session -- entered and exited within this await, in one task, before any
# streaming yield -- to list the tools; each returned tool then opens its own
# per-call session. Holding a persistent session across the streaming response
# (the previous ``async with create_session(...)``) keeps the MCP streamable-http
# client's anyio task group open for the whole run. On the HITL interrupt path the
# response generator is finalized in a different task than it was entered, so that
# task group's cancel scope is exited in the wrong task ("Attempted to exit cancel
# scope in a different task than it was entered in"), which cancels the run and
# drops the buffered interrupt events. Per-call sessions have no long-lived task
# group spanning the stream, so the teardown stays task-local.
raw_tools = await load_mcp_tools(session=None, connection=connection)
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 []
|