fastmcp icon indicating copy to clipboard operation
fastmcp copied to clipboard

Contrib Feature: InterceptedProxyTool

Open strawgate opened this issue 7 months ago • 1 comments

Enhancement Description

I would like to contribute something like the following as a contrib tool (ignore anything specific to my use-case:

class InterceptingProxyTool(ProxyTool):
    """
    A ProxyTool subclass that intercepts the run method to handle redirection.
    """

    additional_input_schema: dict[str, Any] = Field(
        default_factory=lambda: {"type": "object", "properties": {"redirect_to": {"type": "string"}}},
        description="The input schema for the tool.",
    )

    pre_tool_call_hook: Callable[[dict[str, Any], Context | None], None] = Field(
        default_factory=lambda: lambda arguments, context: None, description="A hook that is called before a tool call is made."
    )
    post_tool_call_hook: Callable[[list[TextContent | ImageContent | EmbeddedResource], Context | None], None] = Field(
        default_factory=lambda: lambda response, context: None,  # noqa: ARG005
        description="A hook that is called after a tool call is made.",
    )

You can then setup a multi-server client like so:

        proxy = FastMCP.as_proxy(client)

        proxy_tools: dict[str, FastMCPTool] = await proxy.get_tools()

        for tool in proxy_tools.values():
            new_input_schema = json.loads(json.dumps(tool.parameters))

            add_redirect_to_schema(new_input_schema)

            # Create an instance of our new InterceptingProxyTool
            proxy_tool = InterceptingProxyTool(
                client=client,  # type: ignore
                name=f"oops-{tool.name}",
                description=tool.description,
                parameters=new_input_schema,
                fn=_proxy_passthrough,
                annotations=tool.annotations,
                serializer=tool.serializer,
            )

            mcp._tool_manager.add_tool(proxy_tool)

        await mcp.run_async(transport=mcp_transport)

And then register each "proxied" tool with a new MCP server (ideally i'd do it without accessing the private tool manager).

Use Case

Providing default values for tool calls from certain servers, limiting response sizes from certain servers, adding parameters to tools that you can intercept and act on.

See an example here in my MCPOops server https://github.com/strawgate/py-mcp-collection/blob/main/mcp-oops/src/mcp_oops/main.py

strawgate avatar May 23 '25 13:05 strawgate

#599

strawgate avatar May 26 '25 19:05 strawgate