Coverage for src/ecnl/server.py: 70%

1"""Entry point for the ECNL MCP server. 

2 

3Wires together the outbound adapter (AthleteOne HTTP), discovery, the 

4application service, and the inbound MCP adapter. The dependency graph flows 

5inward — adapters depend on ports, ports depend on domain models. 

6 

7Transport: controlled by the MCP_TRANSPORT environment variable. 

8- "stdio" (default): JSON-RPC over stdin/stdout — client spawns the server as a 

9 subprocess. Never write to stdout in this mode; it corrupts the message stream. 

10- "streamable-http": HTTP server on HOST:PORT. 

11 

12AthleteOne API host: controlled by API_HOST (default: https://api.athleteone.com). 

13 

14Structured logging: all records are emitted as JSON to stderr (captured as 

15container logs). Level via LOG_LEVEL (default: INFO). 

16""" 

17 

18import json 

19import logging 

20import os 

21import sys 

22 

23from dotenv import load_dotenv 

24from mcp.server.fastmcp import FastMCP 

25 

26from .adapters.inbound.authorization import build_authorizer 

27from .adapters.inbound.mcp_adapter import create_mcp_server 

28from .adapters.outbound.athleteone_adapter import AthleteOneAdapter 

29from .adapters.outbound.caching_adapter import CachingAdapter 

30from .adapters.outbound.discovery import DiscoveryAdapter 

31from .adapters.outbound.retry_adapter import RetryingAdapter 

32from .application.service import ECNLService 

33from .observability import setup_tracing 

34from .security import build_token_verifier 

35 

36 

37class _JsonFormatter(logging.Formatter): 

38 """Formats log records as single-line JSON objects.""" 

39 

40 def format(self, record: logging.LogRecord) -> str: 

41 entry: dict = { 

42 "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"), 

43 "level": record.levelname, 

44 "logger": record.name, 

45 "message": record.getMessage(), 

46 } 

47 if record.exc_info: 

48 entry["exception"] = self.formatException(record.exc_info) 

49 return json.dumps(entry) 

50 

51 

52def _configure_logging() -> None: 

53 """Configure the root logger to emit JSON records to stderr (LOG_LEVEL, default INFO).""" 

54 level_name = os.environ.get("LOG_LEVEL", "INFO").upper() 

55 level = getattr(logging, level_name, logging.INFO) 

56 handler = logging.StreamHandler(sys.stderr) 

57 handler.setFormatter(_JsonFormatter()) 

58 logging.root.setLevel(level) 

59 logging.root.addHandler(handler) 

60 

61 

62load_dotenv() 

63_configure_logging() 

64 

65logger = logging.getLogger(__name__) 

66 

67_VALID_TRANSPORTS = ("stdio", "streamable-http") 

68 

69 

70def build_server( 

71 host: str = "0.0.0.0", 

72 port: int = 8000, 

73 api_host: str = "https://api.athleteone.com", 

74 path: str = "/mcp", 

75 auth_settings=None, 

76 token_verifier=None, 

77 authorizer=None, 

78) -> FastMCP: 

79 """Wire AthleteOneAdapter → retry → cache → ECNLService → FastMCP. 

80 

81 Discovery shares the cached adapter, so org/event lookups are cached and 

82 retried just like direct data calls. 

83 

84 Args: 

85 host: Bind address for HTTP transport (ignored for stdio). 

86 port: TCP port for HTTP transport (ignored for stdio). 

87 api_host: Base URL of the upstream AthleteOne API. 

88 path: URL path for the streamable-http transport (ignored for stdio). 

89 auth_settings: Optional AuthSettings to enable bearer-token enforcement. 

90 token_verifier: Optional TokenVerifier consulted on every request. 

91 """ 

92 adapter = AthleteOneAdapter(base_url=api_host) 

93 # Compose cross-cutting adapters: HTTP → retry on transient errors → cache results. 

94 repo = CachingAdapter(RetryingAdapter(adapter)) 

95 discovery = DiscoveryAdapter(repo) 

96 

97 service = ECNLService(repo=repo, discovery=discovery) 

98 return create_mcp_server( 

99 service, 

100 host=host, 

101 port=port, 

102 path=path, 

103 auth_settings=auth_settings, 

104 token_verifier=token_verifier, 

105 authorizer=authorizer, 

106 ) 

107 

108 

109def main() -> None: 

110 """Start the ECNL MCP server. 

111 

112 Reads configuration from the environment: 

113 API_HOST — base URL of the AthleteOne API (default: https://api.athleteone.com) 

114 MCP_TRANSPORT — "stdio" (default) or "streamable-http" 

115 HOST — bind address for HTTP transport (default: 0.0.0.0) 

116 PORT — TCP port for HTTP transport (default: 8000) 

117 MCP_PATH — URL path for streamable-http transport (default: /mcp/ecnl) 

118 MCP_TRACING_ENABLED — bootstrap the OpenTelemetry SDK (default false) 

119 MCP_AUTH_ENABLED — require RS256 bearer tokens on streamable-http (default false) 

120 MCP_AUTH_ISSUER_URL — auth-server origin (required when MCP_AUTH_ENABLED=true) 

121 MCP_AUTH_RESOURCE_URL — this server's public URL for the aud claim (optional) 

122 """ 

123 api_host = os.environ.get("API_HOST", "https://api.athleteone.com") 

124 transport = os.environ.get("MCP_TRANSPORT", "stdio").lower() 

125 host = os.environ.get("HOST", "0.0.0.0") 

126 port = int(os.environ.get("PORT", "8000")) 

127 path = os.environ.get("MCP_PATH", "/mcp/ecnl") 

128 

129 if transport not in _VALID_TRANSPORTS: 

130 raise ValueError(f"Invalid MCP_TRANSPORT={transport!r}. Must be one of: {', '.join(_VALID_TRANSPORTS)}") 

131 

132 auth_settings, token_verifier = _build_auth(transport) 

133 # Build the inbound authorizer once at startup. Defaults to 

134 # PassThroughAuthorizer when MCP_AUTHZ_URL is unset. 

135 authorizer = build_authorizer() 

136 

137 # Wire OpenTelemetry before constructing the server so the 

138 # HTTPXClientInstrumentor patches httpx before any outbound 

139 # adapters are instantiated. A no-op when MCP_TRACING_ENABLED is 

140 # unset; the returned shutdown is invoked on normal exit via 

141 # try/finally so buffered spans flush. 

142 shutdown_tracing = setup_tracing("jk-mcp-ecnl") 

143 try: 

144 if transport == "streamable-http": 

145 logger.info( 

146 "Starting ECNL MCP server (streamable-http transport, %s:%s, path=%s, auth=%s)", 

147 host, 

148 port, 

149 path, 

150 "on" if token_verifier else "off", 

151 ) 

152 build_server( 

153 host=host, 

154 port=port, 

155 api_host=api_host, 

156 path=path, 

157 auth_settings=auth_settings, 

158 token_verifier=token_verifier, 

159 authorizer=authorizer, 

160 ).run(transport="streamable-http") 

161 else: 

162 logger.info("Starting ECNL MCP server (stdio transport)") 

163 build_server(api_host=api_host, authorizer=authorizer).run(transport="stdio") 

164 finally: 

165 shutdown_tracing() 

166 

167 

168def _build_auth(transport: str): 

169 """Build the AuthSettings + TokenVerifier pair for the streamable-http 

170 transport. 

171 

172 The stdio transport relies on the subprocess boundary as its trust 

173 anchor; injecting bearer-token auth there would only confuse 

174 operators. The function returns ``(None, None)`` for stdio 

175 regardless of the env-var state. 

176 """ 

177 if transport != "streamable-http": 

178 return None, None 

179 verifier = build_token_verifier() 

180 if verifier is None: 

181 return None, None 

182 from mcp.server.auth.settings import AuthSettings 

183 

184 settings = AuthSettings( 

185 issuer_url=verifier.issuer, 

186 resource_server_url=verifier.audience or verifier.issuer, 

187 ) 

188 return settings, verifier 

189 

190 

191if __name__ == "__main__": 

192 main()