feat: inject runtime arg for ToolRuntime access (#370)

Theoretically allows for access to tool call id, state, etc when
available in a LG context (like with `create_agent`).

This works w/ the new `langgraph-prebuilt` and `langchain-core`.

```py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain.agents import create_agent
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain_openai import ChatOpenAI
from mcp.types import CallToolResult
from typing_extensions import TypedDict


async def main():

    async def runtime_interceptor(
        request: MCPToolCallRequest, handler
    ) -> CallToolResult:
        print(request.runtime.tool_call_id)
        print(request.runtime.context)
        print(request.runtime.state)
        return await handler(request)

    server_params = StdioServerParameters(
        command="python",
        args=["./math_server.py"],
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await load_mcp_tools(
                session, tool_interceptors=[runtime_interceptor]
            )

            class ContextSchema(TypedDict):
                user_id: str

            agent = create_agent(
                model=ChatOpenAI(model="gpt-4o-mini"),
                tools=tools,
                context_schema=ContextSchema,
            )
            await agent.ainvoke(
                {"messages": [{"role": "user", "content": "What is 5 + 3?"}]},
                context={"user_id": "123"},
            )

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

"""
[11/21/25 10:02:42] INFO     Processing request of type ListToolsRequest                            server.py:674
call_f3Unpu6pMXc3erYy2U540a8s
{'user_id': '123'}
{'messages': [HumanMessage(content='What is 5 + 3?', additional_kwargs={}, response_metadata={}, id='ca0e9e84-34a0-4540-b92f-bcf023316562'), AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 17, 'prompt_tokens': 73, 'total_tokens': 90, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_560af6e559', 'id': 'chatcmpl-CeMwlKviSagQxVwtkocmYe9YzNlRb', 'service_tier': 'default', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--f8fff704-5f76-4526-8dbd-62878b5fbc15-0', tool_calls=[{'name': 'add', 'args': {'a': 5, 'b': 3}, 'id': 'call_f3Unpu6pMXc3erYy2U540a8s', 'type': 'tool_call'}], usage_metadata={'input_tokens': 73, 'output_tokens': 17, 'total_tokens': 90, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})]}
[11/21/25 10:02:44] INFO     Processing request of type CallToolRequest 
"""
```

Author: sydney-runkle

langchain_mcp_adapters/interceptors.py

@@ -12,14 +12,14 @@
 from dataclasses import dataclass, replace
 from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
 
+from langchain_core.messages import ToolMessage
 from mcp.types import CallToolResult
 from typing_extensions import NotRequired, TypedDict, Unpack
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
 
-
-MCPToolCallResult = CallToolResult
+MCPToolCallResult = CallToolResult | ToolMessage
 
 
 class _MCPToolCallRequestOverrides(TypedDict, total=False):

langchain_mcp_adapters/tools.py

@@ -7,6 +7,7 @@
 from collections.abc import Awaitable, Callable
 from typing import Any, get_args
 
+from langchain_core.messages import ToolMessage
 from langchain_core.tools import (
     BaseTool,
     InjectedToolArg,
@@ -35,33 +36,34 @@
 )
 from langchain_mcp_adapters.sessions import Connection, create_session
 
-try:
-    from langgraph.runtime import get_runtime
-except ImportError:
-
-    def get_runtime() -> None:
-        """no-op runtime getter."""
-        return
-
-
 NonTextContent = ImageContent | AudioContent | ResourceLink | EmbeddedResource
 MAX_ITERATIONS = 1000
 
 
 def _convert_call_tool_result(
     call_tool_result: MCPToolCallResult,
-) -> tuple[str | list[str], list[NonTextContent] | None]:
+) -> tuple[str | list[str] | ToolMessage, list[NonTextContent] | None]:
     """Convert MCP MCPToolCallResult to LangChain tool result format.
 
     Args:
-        call_tool_result: The result from calling an MCP tool.
+        call_tool_result: The result from calling an MCP tool. Can be either
+            a CallToolResult (MCP format) or a ToolMessage (LangChain format).
 
     Returns:
-        A tuple containing the text content and any non-text content.
+        A tuple containing the text content (which may be a ToolMessage) and any
+        non-text content. When a ToolMessage is returned by an interceptor, it's
+        placed in the first position of the tuple as the content, with None as
+        the artifact.
 
     Raises:
         ToolException: If the tool call resulted in an error.
     """
+    # If the interceptor returned a ToolMessage directly, return it as the content
+    # with None as the artifact to match the content_and_artifact format
+    if isinstance(call_tool_result, ToolMessage):
+        return call_tool_result, None
+
+    # Otherwise, convert from CallToolResult
     text_contents: list[TextContent] = []
     non_text_contents = []
     for content in call_tool_result.content:
@@ -184,15 +186,18 @@ def convert_mcp_tool_to_langchain_tool(
         raise ValueError(msg)
 
     async def call_tool(
+        runtime: Any = None,  # noqa: ANN401
         **arguments: dict[str, Any],
-    ) -> tuple[str | list[str], list[NonTextContent] | None]:
+    ) -> tuple[str | list[str] | ToolMessage, list[NonTextContent] | None]:
         """Execute tool call with interceptor chain and return formatted result.
 
         Args:
+            runtime: LangGraph tool runtime if available, otherwise None.
             **arguments: Tool arguments as keyword args.
 
         Returns:
-            Tuple of (text_content, non_text_content).
+            A tuple of (text_content, non_text_content), where text_content may be
+            a ToolMessage if an interceptor returned one directly.
         """
         mcp_callbacks = (
             callbacks.to_mcp_format(
@@ -202,12 +207,6 @@ async def call_tool(
             else _MCPCallbacks()
         )
 
-        # try to get runtime if we're in a langgraph context
-        try:
-            runtime = get_runtime()
-        except Exception:  # noqa: BLE001
-            runtime = None
-
         # Create the innermost handler that actually executes the tool call
         async def execute_tool(request: MCPToolCallRequest) -> MCPToolCallResult:
             """Execute the actual MCP tool call with optional session creation.

tests/test_interceptors.py

@@ -1,6 +1,7 @@
 """Tests for the interceptor system functionality."""
 
 import pytest
+from langchain_core.messages import ToolMessage
 from mcp.server import FastMCP
 from mcp.types import (
     CallToolResult,
@@ -283,3 +284,38 @@ async def failing_interceptor(
             add_tool = next(tool for tool in tools if tool.name == "add")
             with pytest.raises(ValueError, match="Interceptor failed"):
                 await add_tool.ainvoke({"a": 2, "b": 3})
+
+    async def test_interceptor_returns_tool_message(self, socket_enabled):
+        """Test that interceptor can return a ToolMessage directly."""
+
+        async def tool_message_interceptor(
+            request: MCPToolCallRequest,
+            handler,
+        ) -> ToolMessage:
+            # Return a ToolMessage directly instead of CallToolResult
+            return ToolMessage(
+                content="Custom ToolMessage response",
+                name=request.name,
+                tool_call_id="test-call-id",
+            )
+
+        with run_streamable_http(_create_math_server, 8209):
+            tools = await load_mcp_tools(
+                None,
+                connection={
+                    "url": "http://localhost:8209/mcp",
+                    "transport": "streamable_http",
+                },
+                tool_interceptors=[tool_message_interceptor],
+            )
+
+            add_tool = next(tool for tool in tools if tool.name == "add")
+            result = await add_tool.ainvoke(
+                {"args": {"a": 2, "b": 3}, "id": "test-call-id", "type": "tool_call"}
+            )
+
+            # The interceptor returns a ToolMessage which should be returned as-is
+            assert isinstance(result, ToolMessage)
+            assert result.content == "Custom ToolMessage response"
+            assert result.name == "add"
+            assert result.tool_call_id == "test-call-id"