Coverage for src/epl/server.py: 71%
69 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-05 10:11 +0000
1"""Entry point for the Premier 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.pulselive_adapter import PulseliveAdapter
36from .adapters.outbound.retry_adapter import RetryingAdapter
37from .application.service import EPLService
38from .observability import setup_tracing
39from .security import build_token_verifier
41_PULSELIVE_ADAPTER = PulseliveAdapter
44class _JsonFormatter(logging.Formatter):
45 """Formats log records as single-line JSON objects.
47 Using a custom formatter rather than a third-party library keeps the
48 dependency surface minimal. Each record becomes one JSON line, which is
49 the format expected by most container log drivers and aggregators.
50 """
52 def format(self, record: logging.LogRecord) -> str:
53 entry: dict = {
54 "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
55 "level": record.levelname,
56 "logger": record.name,
57 "message": record.getMessage(),
58 }
59 if record.exc_info:
60 entry["exception"] = self.formatException(record.exc_info)
61 return json.dumps(entry)
64def _configure_logging() -> None:
65 """Configure the root logger to emit JSON records to stderr.
67 Reads the LOG_LEVEL environment variable (default: INFO). Invalid values
68 fall back to INFO.
69 """
70 level_name = os.environ.get("LOG_LEVEL", "INFO").upper()
71 level = getattr(logging, level_name, logging.INFO)
72 handler = logging.StreamHandler(sys.stderr)
73 handler.setFormatter(_JsonFormatter())
74 logging.root.setLevel(level)
75 logging.root.addHandler(handler)
78load_dotenv()
79_configure_logging()
81logger = logging.getLogger(__name__)
83_VALID_TRANSPORTS = ("stdio", "streamable-http")
86def build_server(
87 host: str = "0.0.0.0",
88 port: int = 8000,
89 api_host: str = "https://site.api.espn.com",
90 path: str = "/mcp",
91 auth_settings=None,
92 token_verifier=None,
93 authorizer=None,
94) -> FastMCP:
95 """Wire ESPNAdapter → EPLService → FastMCP and return the server.
97 Extracted from main() so the composition graph can be constructed and
98 tested without starting any transport.
100 Args:
101 host: Bind address for HTTP transport (ignored for stdio). Defaults to 0.0.0.0.
102 port: TCP port for HTTP transport (ignored for stdio). Defaults to 8000.
103 api_host: Base URL of the upstream ESPN API.
104 path: URL path for the streamable-http transport (ignored for stdio).
105 auth_settings: Optional AuthSettings to enable bearer-token enforcement.
106 token_verifier: Optional TokenVerifier consulted on every request.
107 """
108 adapter = ESPNAdapter(base_url=api_host)
109 # Compose cross-cutting adapters: HTTP → retry on transient errors → cache results.
110 retrying = RetryingAdapter(adapter)
111 caching = CachingAdapter(retrying)
113 pulselive = _PULSELIVE_ADAPTER()
115 service = EPLService(repo=caching, pulselive=pulselive)
116 return create_mcp_server(
117 service,
118 host=host,
119 port=port,
120 path=path,
121 auth_settings=auth_settings,
122 token_verifier=token_verifier,
123 authorizer=authorizer,
124 )
127def main() -> None:
128 """Start the Premier League MCP server.
130 Reads configuration from the environment:
131 API_HOST — base URL of the upstream ESPN API (default: https://site.api.espn.com)
132 MCP_TRANSPORT — "stdio" (default) or "streamable-http"
133 HOST — bind address for HTTP transport (default: 0.0.0.0)
134 PORT — TCP port for HTTP transport (default: 8000)
135 MCP_PATH — URL path for streamable-http transport (default: /mcp/epl)
136 MCP_TRACING_ENABLED — bootstrap the OpenTelemetry SDK (default false)
137 MCP_AUTH_ENABLED — require RS256 bearer tokens on streamable-http (default false)
138 MCP_AUTH_ISSUER_URL — auth-server origin (required when MCP_AUTH_ENABLED=true)
139 MCP_AUTH_RESOURCE_URL — this server's public URL for the aud claim (optional)
140 """
141 api_host = os.environ.get("API_HOST", "https://site.api.espn.com")
142 transport = os.environ.get("MCP_TRANSPORT", "stdio").lower()
143 host = os.environ.get("HOST", "0.0.0.0")
144 port = int(os.environ.get("PORT", "8000"))
145 path = os.environ.get("MCP_PATH", "/mcp/epl")
147 if transport not in _VALID_TRANSPORTS:
148 raise ValueError(f"Invalid MCP_TRANSPORT={transport!r}. Must be one of: {', '.join(_VALID_TRANSPORTS)}")
150 auth_settings, token_verifier = _build_auth(transport)
151 # Build the inbound authorizer once at startup. Defaults to
152 # PassThroughAuthorizer when MCP_AUTHZ_URL is unset, matching the
153 # local-dev and stdio posture; production deployments set the URL
154 # so every tool dispatch consults authorization-policy-service.
155 authorizer = build_authorizer()
157 # Wire OpenTelemetry before constructing the server so the
158 # HTTPXClientInstrumentor patches httpx before any outbound
159 # adapters are instantiated. A no-op when MCP_TRACING_ENABLED is
160 # unset; the returned shutdown is invoked on normal exit via
161 # try/finally so buffered spans flush.
162 shutdown_tracing = setup_tracing("jk-mcp-epl")
163 try:
164 if transport == "streamable-http":
165 logger.info(
166 "Starting Premier League MCP server (streamable-http transport, %s:%s, path=%s, auth=%s)",
167 host,
168 port,
169 path,
170 "on" if token_verifier else "off",
171 )
172 build_server(
173 host=host,
174 port=port,
175 api_host=api_host,
176 path=path,
177 auth_settings=auth_settings,
178 token_verifier=token_verifier,
179 authorizer=authorizer,
180 ).run(transport="streamable-http")
181 else:
182 logger.info("Starting Premier League MCP server (stdio transport)")
183 build_server(api_host=api_host, authorizer=authorizer).run(transport="stdio")
184 finally:
185 shutdown_tracing()
188def _build_auth(transport: str):
189 """Build the AuthSettings + TokenVerifier pair for the streamable-http
190 transport.
192 The stdio transport relies on the subprocess boundary as its trust
193 anchor; injecting bearer-token auth there would only confuse
194 operators. The function returns ``(None, None)`` for stdio
195 regardless of the env-var state.
196 """
197 if transport != "streamable-http":
198 return None, None
199 verifier = build_token_verifier()
200 if verifier is None:
201 return None, None
202 # AuthSettings is imported lazily — when MCP_AUTH_ENABLED is unset
203 # the import cost is skipped, matching the same lazy posture the
204 # observability bootstrap uses.
205 from mcp.server.auth.settings import AuthSettings
207 settings = AuthSettings(
208 issuer_url=verifier.issuer,
209 resource_server_url=verifier.audience or verifier.issuer,
210 )
211 return settings, verifier
214if __name__ == "__main__":
215 main()