Description
Hi there,
I am running into an issue when attempting to use callable objects with FastMCP.
Description / Reproducible Example
Regular, synchronous callable objects work (mostly) as expected, with the exception of having to set self.__name__
:
import asyncio
from mcp.server.fastmcp import FastMCP
class TestTool:
def __init__(self) -> None:
self.__name__ = "TestTool"
def __call__(self) -> str:
return "Hello from TestTool!"
mcp = FastMCP()
mcp.add_tool(TestTool())
async def main() -> None:
# prints "Hello from TestTool!"
print(await mcp.call_tool("TestTool", {}))
if __name__ == "__main__":
asyncio.run(main())
However, making the __call__
method async breaks things:
import asyncio
from mcp.server.fastmcp import FastMCP
class TestTool:
def __init__(self) -> None:
self.__name__ = "TestTool"
async def __call__(self) -> str:
return "Hello from TestTool!"
mcp = FastMCP()
mcp.add_tool(TestTool())
async def main() -> None:
# prints [TextContent(type='text', text='<coroutine object TestTool.__call__ at 0x103155300>', annotations=None)]
print(await mcp.call_tool("TestTool", {}))
if __name__ == "__main__":
asyncio.run(main())
Simple Fix
I tracked this down to this line in Tool.from_function
, which does not work properly for async callable objects:
is_async = inspect.iscoroutinefunction(fn)
I know this use-case is slightly out-of-the-ordinary, but it would be extremely helpful if we could improve this to work properly for callable objects as well.
There is some prior art here, Starlette checks if the provided callable is async like this, which works for both regular async functions and async callable objects:
def is_async_callable(obj: typing.Any) -> typing.Any:
while isinstance(obj, functools.partial):
obj = obj.func
return inspect.iscoroutinefunction(obj) or (callable(obj) and inspect.iscoroutinefunction(obj.__call__))
Would you be open to accepting a pull request with a similar implementation?
Thank you!