Module fastmcp_extensions.utils.test_tool

MCP tool testing utilities.

This module provides utilities for testing MCP tools directly with JSON arguments, supporting both stdio and HTTP transports.

Usage (stdio transport): python -m fastmcp_extensions.utils.test_tool –app ''

Example:
    python -m fastmcp_extensions.utils.test_tool --app my_mcp_server.server:app list_tools '{}'

Poe task configuration:
    [tool.poe.tasks.mcp-tool-test]
    cmd = "python -m fastmcp_extensions.utils.test_tool --app my_mcp_server.server:app"
    help = "Test MCP tools with JSON arguments"

Usage (HTTP transport): python -m fastmcp_extensions.utils.test_tool –http –app [tool_name] ['']

Example:
    python -m fastmcp_extensions.utils.test_tool --http --app my_mcp_server.server:app
    python -m fastmcp_extensions.utils.test_tool --http --app my_mcp_server.server:app get_version '{}'

Poe task configuration:
    [tool.poe.tasks.mcp-tool-test-http]
    cmd = "python -m fastmcp_extensions.utils.test_tool --http --app my_mcp_server.server:app"
    help = "Test MCP tools over HTTP transport"

Functions

async def call_mcp_tool(app: FastMCP, tool_name: str, args: dict[str, Any]) ‑> object

Expand source code

async def call_mcp_tool(app: FastMCP, tool_name: str, args: dict[str, Any]) -> object:
    """Call an MCP tool using the FastMCP client."""
    async with Client(app) as client:
        return await client.call_tool(tool_name, args)

Call an MCP tool using the FastMCP client.

def find_free_port() ‑> int

Expand source code

def find_free_port() -> int:
    """Find an available port on localhost."""
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]

Find an available port on localhost.

async def list_mcp_tools(app: FastMCP) ‑> list[Any]

Expand source code

async def list_mcp_tools(app: FastMCP) -> list[Any]:
    """List all available MCP tools."""
    async with Client(app) as client:
        return await client.list_tools()

List all available MCP tools.

async def run_http_tool_test(app: FastMCP,
port: int | None = None,
tool_name: str | None = None,
args: dict[str, Any] | None = None) ‑> int

Expand source code

async def run_http_tool_test(
    app: FastMCP,
    port: int | None = None,
    tool_name: str | None = None,
    args: dict[str, Any] | None = None,
) -> int:
    """Run a tool test over HTTP transport using the app directly."""
    import uvicorn

    if port is None:
        port = find_free_port()

    url = f"http://127.0.0.1:{port}/mcp"
    os.environ["MCP_HTTP_PORT"] = str(port)

    print(f"Starting HTTP server on port {port}...", file=sys.stderr)

    server_error: Exception | None = None

    def run_server() -> None:
        nonlocal server_error
        try:
            uvicorn.run(
                app.http_app(),
                host="127.0.0.1",
                port=port,
                log_level="error",
            )
        except Exception as e:
            server_error = e

    server_thread = threading.Thread(target=run_server, daemon=True)
    server_thread.start()

    try:
        if not await wait_for_server(url):
            if server_error:
                print(f"Server error: {server_error}", file=sys.stderr)
            print(f"Server failed to start on port {port}", file=sys.stderr)
            return 1

        async with Client(url) as client:
            tools = await client.list_tools()
            print(f"HTTP transport OK - {len(tools)} tools available")

            if tool_name:
                print(f"Calling tool: {tool_name}", file=sys.stderr)
                result = await client.call_tool(tool_name, args or {})

                if hasattr(result, "text"):
                    print(result.text)
                else:
                    print(str(result))

        return 0

    finally:
        pass

Run a tool test over HTTP transport using the app directly.

def run_tool_test(app: FastMCP, tool_name: str, json_args: str) ‑> None

Expand source code

def run_tool_test(
    app: FastMCP,
    tool_name: str,
    json_args: str,
) -> None:
    """Run a tool test with JSON arguments and print the result."""
    args: dict[str, Any] = json.loads(json_args)
    result = asyncio.run(call_mcp_tool(app, tool_name, args))

    if hasattr(result, "text"):
        print(result.text)
    else:
        print(str(result))

Run a tool test with JSON arguments and print the result.

async def wait_for_server(url: str, timeout: float = 10.0) ‑> bool

Expand source code

async def wait_for_server(url: str, timeout: float = SERVER_STARTUP_TIMEOUT) -> bool:
    """Wait for the MCP server to be ready by attempting to list tools."""
    deadline = asyncio.get_event_loop().time() + timeout
    while asyncio.get_event_loop().time() < deadline:
        try:
            async with Client(url) as client:
                await client.list_tools()
                return True
        except Exception:
            await asyncio.sleep(POLL_INTERVAL)
    return False

Wait for the MCP server to be ready by attempting to list tools.