Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion src/mcp/server/mcpserver/utilities/context_injection.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,17 @@ def find_context_parameter(fn: Callable[..., Any]) -> str | None:
"""
from mcp.server.mcpserver.server import Context

# Handle callable class instances by using __call__ method,
# since typing.get_type_hints() doesn't introspect __call__
# on class instances.
target = fn
if not (inspect.isfunction(fn) or inspect.ismethod(fn)):
if callable(fn) and hasattr(fn, "__call__"):
target = fn.__call__

# Get type hints to properly resolve string annotations
try:
hints = typing.get_type_hints(fn)
hints = typing.get_type_hints(target)
except Exception: # pragma: lax no cover
# If we can't resolve type hints, we can't find the context parameter
return None
Expand Down
52 changes: 52 additions & 0 deletions tests/server/mcpserver/test_tool_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,58 @@ def tool_with_context(x: int, ctx: Context[ServerSessionT, None]) -> str:
with pytest.raises(ToolError, match="Error executing tool tool_with_context"):
await manager.call_tool("tool_with_context", {"x": 42}, context=ctx)

def test_context_parameter_detection_callable_class(self):
"""Test that context parameters are detected in callable class instances."""

class MyTool:
def __init__(self):
self.__name__ = "my_tool"

def __call__(self, x: int, ctx: Context[ServerSessionT, None]) -> str: # pragma: no cover
return str(x)

manager = ToolManager()
tool = manager.add_tool(MyTool())
assert tool.context_kwarg == "ctx"
assert "ctx" not in json.dumps(tool.parameters)
assert "Context" not in json.dumps(tool.parameters)

def test_context_parameter_detection_async_callable_class(self):
"""Test that context parameters are detected in async callable class instances."""

class MyAsyncTool:
def __init__(self):
self.__name__ = "my_async_tool"

async def __call__(self, x: int, ctx: Context[ServerSessionT, None]) -> str: # pragma: no cover
return str(x)

manager = ToolManager()
tool = manager.add_tool(MyAsyncTool())
assert tool.context_kwarg == "ctx"
assert tool.is_async is True
assert "ctx" not in json.dumps(tool.parameters)

@pytest.mark.anyio
async def test_context_injection_callable_class(self):
"""Test that context is properly injected in callable class tools."""

class MyTool:
def __init__(self):
self.__name__ = "my_tool"

def __call__(self, x: int, ctx: Context[ServerSessionT, None]) -> str:
assert isinstance(ctx, Context)
return str(x)

manager = ToolManager()
manager.add_tool(MyTool())

mcp = MCPServer()
ctx = mcp.get_context()
result = await manager.call_tool("my_tool", {"x": 42}, context=ctx)
assert result == "42"


class TestToolAnnotations:
def test_tool_annotations(self):
Expand Down
Loading