Coverage for src/usls/server.py: 70%
66 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-05 10:11 +0000
1"""Entry point for the USL Super League MCP server.
3Wires together the outbound adapter (ESPN HTTP), the application service, and
4the inbound MCP adapter. The dependency graph flows inward — adapters depend on
5ports, ports depend on domain models. Nothing here is circular.
7Transport: controlled by the MCP_TRANSPORT environment variable.
8- "stdio" (default): JSON-RPC over stdin/stdout — client spawns the server as
9 a subprocess. Never write to stdout in this mode; it corrupts the message stream.
10- "streamable-http": HTTP server on HOST:PORT. The MCP client connects over HTTP.
11 Use this for deployed / networked deployments. HOST defaults to "0.0.0.0" and
12 PORT defaults to 8000.
14ESPN API host: controlled by the API_HOST environment variable.
15- Default: https://site.api.espn.com
17Structured logging: all log records are emitted as JSON objects to stderr.
18Docker captures stderr as container logs, so JSON output is parseable by
19log aggregators without additional parsing rules. The log level can be
20overridden at runtime with the LOG_LEVEL environment variable (default: INFO).
21"""
23import json
24import logging
25import os
26import sys
28from dotenv import load_dotenv
29from mcp.server.fastmcp import FastMCP
31from .adapters.inbound.authorization import build_authorizer
32from .adapters.inbound.mcp_adapter import create_mcp_server
33from .adapters.outbound.caching_adapter import CachingAdapter
34from .adapters.outbound.espn_adapter import ESPNAdapter
35from .adapters.outbound.retry_adapter import RetryingAdapter
36from .application.service import USLSService
37from .observability import setup_tracing
38from .security import build_token_verifier
41class _JsonFormatter(logging.Formatter):
42 """Formats log records as single-line JSON objects.
44 Using a custom formatter rather than a third-party library keeps the
45 dependency surface minimal. Each record becomes one JSON line, which is
46 the format expected by most container log drivers and aggregators.
47 """
49 def format(self, record: logging.LogRecord) -> str:
50 entry: dict = {
51 "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
52 "level": record.levelname,
53 "logger": record.name,
54 "message": record.getMessage(),
55 }
56 if record.exc_info:
57 entry["exception"] = self.formatException(record.exc_info)
58 return json.dumps(entry)
61def _configure_logging() -> None:
62 """Configure the root logger to emit JSON records to stderr.
64 Reads the LOG_LEVEL environment variable (default: INFO). Invalid values
65 fall back to INFO.
66 """
67 level_name = os.environ.get("LOG_LEVEL", "INFO").upper()
68 level = getattr(logging, level_name, logging.INFO)
69 handler = logging.StreamHandler(sys.stderr)
70 handler.setFormatter(_JsonFormatter())
71 logging.root.setLevel(level)
72 logging.root.addHandler(handler)
75load_dotenv()
76_configure_logging()
78logger = logging.getLogger(__name__)
80_VALID_TRANSPORTS = ("stdio", "streamable-http")
83def build_server(
84 host: str = "0.0.0.0",
85 port: int = 8000,
86 api_host: str = "https://site.api.espn.com",
87 path: str = "/mcp",
88 auth_settings=None,
89 token_verifier=None,
90 authorizer=None,
91) -> FastMCP:
92 """Wire ESPNAdapter → USLSService → FastMCP and return the server.
94 Extracted from main() so the composition graph can be constructed and
95 tested without starting any transport.
97 Args:
98 host: Bind address for HTTP transport (ignored for stdio). Defaults to 0.0.0.0.
99 port: TCP port for HTTP transport (ignored for stdio). Defaults to 8000.
100 api_host: Base URL of the upstream ESPN API.
101 path: URL path for the streamable-http transport (ignored for stdio).
102 auth_settings: Optional AuthSettings to enable bearer-token enforcement.
103 token_verifier: Optional TokenVerifier consulted on every request.
104 """
105 adapter = ESPNAdapter(base_url=api_host)
106 # Compose cross-cutting adapters: HTTP → retry on transient errors → cache results.
107 retrying = RetryingAdapter(adapter)
108 caching = CachingAdapter(retrying)
110 service = USLSService(repo=caching)
111 return create_mcp_server(
112 service,
113 host=host,
114 port=port,
115 path=path,
116 auth_settings=auth_settings,
117 token_verifier=token_verifier,
118 authorizer=authorizer,
119 )
122def main() -> None:
123 """Start the USL Super League MCP server.
125 Reads configuration from the environment:
126 API_HOST — base URL of the upstream ESPN API (default: https://site.api.espn.com)
127 MCP_TRANSPORT — "stdio" (default) or "streamable-http"
128 HOST — bind address for HTTP transport (default: 0.0.0.0)
129 PORT — TCP port for HTTP transport (default: 8000)
130 MCP_PATH — URL path for streamable-http transport (default: /mcp/usls)
131 MCP_TRACING_ENABLED — bootstrap the OpenTelemetry SDK (default false)
132 MCP_AUTH_ENABLED — require RS256 bearer tokens on streamable-http (default false)
133 MCP_AUTH_ISSUER_URL — auth-server origin (required when MCP_AUTH_ENABLED=true)
134 MCP_AUTH_RESOURCE_URL — this server's public URL for the aud claim (optional)
135 """
136 api_host = os.environ.get("API_HOST", "https://site.api.espn.com")
137 transport = os.environ.get("MCP_TRANSPORT", "stdio").lower()
138 host = os.environ.get("HOST", "0.0.0.0")
139 port = int(os.environ.get("PORT", "8000"))
140 path = os.environ.get("MCP_PATH", "/mcp/usls")
142 if transport not in _VALID_TRANSPORTS:
143 raise ValueError(f"Invalid MCP_TRANSPORT={transport!r}. Must be one of: {', '.join(_VALID_TRANSPORTS)}")
145 auth_settings, token_verifier = _build_auth(transport)
146 # Build the inbound authorizer once at startup. Defaults to
147 # PassThroughAuthorizer when MCP_AUTHZ_URL is unset, matching the
148 # local-dev and stdio posture; production deployments set the URL
149 # so every tool dispatch consults authorization-policy-service.
150 authorizer = build_authorizer()
152 # Wire OpenTelemetry before constructing the server so the
153 # HTTPXClientInstrumentor patches httpx before any outbound
154 # adapters are instantiated. A no-op when MCP_TRACING_ENABLED is
155 # unset; the returned shutdown is invoked on normal exit via
156 # try/finally so buffered spans flush.
157 shutdown_tracing = setup_tracing("jk-mcp-usls")
158 try:
159 if transport == "streamable-http":
160 logger.info(
161 "Starting USL Super League MCP server (streamable-http transport, %s:%s, path=%s, auth=%s)",
162 host,
163 port,
164 path,
165 "on" if token_verifier else "off",
166 )
167 build_server(
168 host=host,
169 port=port,
170 api_host=api_host,
171 path=path,
172 auth_settings=auth_settings,
173 token_verifier=token_verifier,
174 authorizer=authorizer,
175 ).run(transport="streamable-http")
176 else:
177 logger.info("Starting USL Super League MCP server (stdio transport)")
178 build_server(api_host=api_host, authorizer=authorizer).run(transport="stdio")
179 finally:
180 shutdown_tracing()
183def _build_auth(transport: str):
184 """Build the AuthSettings + TokenVerifier pair for the streamable-http
185 transport.
187 The stdio transport relies on the subprocess boundary as its trust
188 anchor; injecting bearer-token auth there would only confuse
189 operators. The function returns ``(None, None)`` for stdio
190 regardless of the env-var state.
191 """
192 if transport != "streamable-http":
193 return None, None
194 verifier = build_token_verifier()
195 if verifier is None:
196 return None, None
197 # AuthSettings is imported lazily — when MCP_AUTH_ENABLED is unset
198 # the import cost is skipped, matching the same lazy posture the
199 # observability bootstrap uses.
200 from mcp.server.auth.settings import AuthSettings
202 settings = AuthSettings(
203 issuer_url=verifier.issuer,
204 resource_server_url=verifier.audience or verifier.issuer,
205 )
206 return settings, verifier
209if __name__ == "__main__":
210 main()