(settings: config.Settings)
| 54 | |
| 55 | |
| 56 | def create_server(settings: config.Settings) -> FastMCP[None]: |
| 57 | # OAuth discovery is the credential fallback for HTTP transport: only when |
| 58 | # the server holds no static token does it advertise the AS and gate on a |
| 59 | # missing Authorization header. Otherwise (stdio, static token, or a |
| 60 | # forwarded --header) it's pure pass-through. |
| 61 | auth = None |
| 62 | if settings.transport == "http" and settings.flagsmith_api_token is None: |
| 63 | auth = FlagsmithResourceAuth( |
| 64 | resource_url=str(settings.mcp_server_url), |
| 65 | authorization_server=str(settings.flagsmith_api_url), |
| 66 | ) |
| 67 | api_client = httpx.AsyncClient( |
| 68 | base_url=str(settings.flagsmith_api_url), |
| 69 | auth=FlagsmithAuth(settings.flagsmith_api_token), |
| 70 | event_hooks={"request": [propagate_span_attributes]}, |
| 71 | ) |
| 72 | # Instrument only the Flagsmith API client: emit a span per upstream |
| 73 | # call and propagate W3C trace context; the event hook passes the MCP |
| 74 | # call context to the API as W3C Baggage. |
| 75 | HTTPXClientInstrumentor().instrument_client(api_client) |
| 76 | server = FastMCP.from_openapi( |
| 77 | openapi_spec=_fetch_spec(), |
| 78 | client=api_client, |
| 79 | name="Flagsmith", |
| 80 | route_maps=ROUTE_MAPS, |
| 81 | mcp_component_fn=_customise, |
| 82 | validate_output=False, # TODO https://github.com/Flagsmith/flagsmith/issues/7679 |
| 83 | auth=auth, |
| 84 | ) |
| 85 | |
| 86 | server.add_middleware(PrometheusMiddleware()) |
| 87 | server.add_middleware(EventLoggingMiddleware()) |
| 88 | |
| 89 | @server.custom_route("/health", methods=["GET"]) |
| 90 | async def health(request: Request) -> PlainTextResponse: |
| 91 | return PlainTextResponse("OK") |
| 92 | |
| 93 | return server |
| 94 | |
| 95 | |
| 96 | def run() -> None: |
no test coverage detected