OpenAPI 模块¶

def add_tools_endpoint(app: "FastAPI", route_table: RouteTable) -> None:
    """Add a /tools endpoint that returns the list of available tools.

    This endpoint supports ETag-based cache validation. The response includes
    the current ETag in both the response body and headers.

    Args:
        app: The FastAPI application instance.
        route_table: The RouteTable to query for tools.

    Example:
        ```python
        from fastapi import FastAPI
        from toolregistry_server import RouteTable
        from toolregistry_server.openapi.adapter import add_tools_endpoint

        app = FastAPI()
        route_table = RouteTable(registry)
        add_tools_endpoint(app, route_table)
        ```
    """
    try:
        from fastapi import Request
        from fastapi.responses import JSONResponse
    except ImportError as e:
        raise ImportError(
            "FastAPI is required for OpenAPI support. "
            "Install with: pip install toolregistry-server[openapi]"
        ) from e

    @app.get("/tools", tags=["meta"])
    async def list_tools(request: Request):
        """List all available tools with their metadata.

        Returns a JSON object containing:
        - tools: List of tool information (name, namespace, method, path, description)
        - etag: Current ETag for cache validation

        Supports conditional requests via If-None-Match header.
        Returns 304 Not Modified if the ETag matches.
        """
        # Check If-None-Match header
        if_none_match = request.headers.get("If-None-Match")
        current_etag = route_table.etag

        if if_none_match and if_none_match == current_etag:
            return JSONResponse(
                content=None,
                status_code=304,
                headers={"ETag": current_etag},
            )

        # Build tools list — deferred tools are excluded from the initial
        # listing so that LLMs discover them via discover_tools.
        tools = []
        for route in route_table.list_routes(enabled_only=True, include_deferred=False):
            tools.append(
                {
                    "name": route.tool_name,
                    "namespace": route.namespace,
                    "method": route.method_name,
                    "path": route.path,
                    "description": route.description,
                }
            )

        return JSONResponse(
            content={"tools": tools, "etag": current_etag},
            headers={"ETag": current_etag},
        )

def route_table_to_router(
    route_table: RouteTable,
    prefix: str = "",
) -> "APIRouter":
    """Convert a :class:`~toolregistry_server.RouteTable` into a FastAPI router.

    Routes are generated for **all** registered tools regardless of their
    current enabled/disabled state. Each endpoint checks the route's enabled
    state at request time and returns HTTP 503 if the tool has been disabled,
    allowing runtime enable/disable without restarting the server.

    Args:
        route_table: The route table to convert.
        prefix: URL prefix for all generated routes.

    Returns:
        A :class:`~fastapi.APIRouter` with one POST route per registered tool.

    Raises:
        ImportError: If FastAPI is not installed.
    """
    try:
        from fastapi import APIRouter
    except ImportError as e:
        raise ImportError(
            "FastAPI is required for OpenAPI support. "
            "Install with: pip install toolregistry-server[openapi]"
        ) from e

    router = APIRouter(prefix=prefix)

    for route in route_table.list_routes(enabled_only=False):
        _add_route_from_entry(router, route, route_table)

    return router

def setup_dynamic_openapi(app: "FastAPI", route_table: RouteTable) -> None:
    """Configure dynamic OpenAPI schema generation that filters out disabled tools.

    This replaces FastAPI's default cached OpenAPI schema with a dynamic one
    that checks tool enable/disable status on every request to ``/openapi.json``.
    Disabled tools are excluded from the schema so they do not appear in
    ``/docs`` or ``/openapi.json``, and re-enabling them makes them visible
    again immediately.

    Args:
        app: The FastAPI application instance.
        route_table: The route table used for enable/disable status checks.
    """
    try:
        from fastapi.openapi.utils import get_openapi
    except ImportError as e:
        raise ImportError(
            "FastAPI is required for OpenAPI support. "
            "Install with: pip install toolregistry-server[openapi]"
        ) from e

    def custom_openapi() -> dict[str, Any]:
        # Generate a fresh OpenAPI schema on every call (no caching)
        # so it always reflects the current enable/disable state.
        openapi_schema = get_openapi(
            title=app.title,
            version=app.version,
            description=app.description,
            routes=app.routes,
        )

        # Collect operation_ids of disabled tools
        disabled_operation_ids: set[str] = set()
        for route in route_table.list_routes(enabled_only=False):
            if not route.enabled:
                disabled_operation_ids.add(route.tool_name)

        # Filter out paths whose operations correspond to disabled tools
        if disabled_operation_ids and "paths" in openapi_schema:
            openapi_schema["paths"] = _filter_disabled_paths(
                openapi_schema["paths"], disabled_operation_ids
            )

        # Do NOT cache (app.openapi_schema is not set) so the schema
        # is regenerated on every request, reflecting runtime changes.
        return openapi_schema

    app.openapi = custom_openapi  # ty: ignore[invalid-assignment]

def add_etag_middleware(app: FastAPI, route_table: RouteTable) -> None:
    """Add ETag middleware to a FastAPI application.

    This is a convenience function to add the ETagMiddleware to a FastAPI
    application with the correct configuration.

    Args:
        app: The FastAPI application instance.
        route_table: The RouteTable instance for ETag generation.

    Example:
        ```python
        from fastapi import FastAPI
        from toolregistry_server import RouteTable
        from toolregistry_server.openapi.middleware import add_etag_middleware

        app = FastAPI()
        route_table = RouteTable(registry)
        add_etag_middleware(app, route_table)
        ```
    """
    app.add_middleware(ETagMiddleware, route_table=route_table)  # ty: ignore[invalid-argument-type]