Module fastmcp_extensions.utils

FastMCP Extensions utilities for testing and server description.

This module contains utilities that are designed to be called as scripts or used programmatically for testing and describing MCP servers.

Submodules

• test_tool: MCP tool testing utilities (stdio and HTTP transports)
• describe_server: MCP server description and measurement utilities

Sub-modules

fastmcp_extensions.utils.describe_server

MCP server description and measurement utilities …

fastmcp_extensions.utils.test_tool

MCP tool testing utilities …

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 get_tool_details(app: FastMCP) ‑> list[dict[str, Any]]

Expand source code

async def get_tool_details(app: FastMCP) -> list[dict[str, Any]]:
    """Get detailed information about each tool.

    Args:
        app: The FastMCP app instance

    Returns:
        List of dictionaries with tool details including name, description length,
        and schema length for each tool.
    """
    async with Client(app) as client:
        tools = await client.list_tools()

        details = []
        for tool in tools:
            name_len = len(tool.name)
            desc_len = len(tool.description) if tool.description else 0
            schema_len = len(str(tool.inputSchema)) if tool.inputSchema else 0

            details.append(
                {
                    "name": tool.name,
                    "name_length": name_len,
                    "description_length": desc_len,
                    "schema_length": schema_len,
                    "total_length": name_len + desc_len + schema_len,
                }
            )

        return details

Get detailed information about each tool.

Args

app

The FastMCP app instance

Returns

List of dictionaries with tool details including name, description length, and schema length for each tool.

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 measure_tool_list(app: FastMCP) ‑> tuple[int, int]

Expand source code

async def measure_tool_list(app: FastMCP) -> tuple[int, int]:
    """Measure the tool list size from the MCP server.

    This function connects to the MCP server and measures the character count
    of the tool list, including tool names, descriptions, and input schemas.

    Args:
        app: The FastMCP app instance

    Returns:
        Tuple of (tool_count, total_character_count)
    """
    async with Client(app) as client:
        tools = await client.list_tools()

        tool_count = len(tools)
        total_chars = 0

        for tool in tools:
            total_chars += len(tool.name)

            if tool.description:
                total_chars += len(tool.description)

            if tool.inputSchema:
                total_chars += len(str(tool.inputSchema))

        return tool_count, total_chars

Measure the tool list size from the MCP server.

This function connects to the MCP server and measures the character count of the tool list, including tool names, descriptions, and input schemas.

Args

app

The FastMCP app instance

Returns

Tuple of (tool_count, total_character_count)

async def measure_tool_list_detailed(app: FastMCP, server_name: str | None = None) ‑> ToolListMeasurement

Expand source code

async def measure_tool_list_detailed(
    app: FastMCP,
    server_name: str | None = None,
) -> ToolListMeasurement:
    """Measure the tool list size with detailed results.

    Args:
        app: The FastMCP app instance
        server_name: Optional name of the server for reporting

    Returns:
        ToolListMeasurement with detailed results
    """
    tool_count, total_chars = await measure_tool_list(app)

    return ToolListMeasurement(
        tool_count=tool_count,
        total_characters=total_chars,
        average_chars_per_tool=total_chars // tool_count if tool_count > 0 else 0,
        server_name=server_name,
    )

Measure the tool list size with detailed results.

Args

app

The FastMCP app instance

server_name

Optional name of the server for reporting

Returns

ToolListMeasurement with detailed results

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_measurement(app: FastMCP, server_name: str | None = None) ‑> None

Expand source code

def run_measurement(app: FastMCP, server_name: str | None = None) -> None:
    """Run tool list measurement and print results.

    This is a convenience function for CLI measurement scripts.

    Args:
        app: The FastMCP app instance
        server_name: Optional name of the server for reporting
    """
    measurement = asyncio.run(measure_tool_list_detailed(app, server_name))
    print(str(measurement))

Run tool list measurement and print results.

This is a convenience function for CLI measurement scripts.

Args

app

The FastMCP app instance

server_name

Optional name of the server for reporting

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.

Classes

class ToolListMeasurement (tool_count: int,
total_characters: int,
average_chars_per_tool: int,
server_name: str | None = None)

Expand source code

@dataclass
class ToolListMeasurement:
    """Measurement results for an MCP tool list."""

    tool_count: int
    total_characters: int
    average_chars_per_tool: int
    server_name: str | None = None

    def __str__(self) -> str:
        """Return a human-readable string representation."""
        lines = []
        if self.server_name:
            lines.append(f"MCP Server: {self.server_name}")
        lines.append(f"Tool count: {self.tool_count}")
        lines.append(f"Total characters: {self.total_characters:,}")
        lines.append(f"Average chars per tool: {self.average_chars_per_tool:,}")
        return "\n".join(lines)

Measurement results for an MCP tool list.

Instance variables

var average_chars_per_tool : int

var server_name : str | None

var tool_count : int

var total_characters : int