| 1 | """Official MCP protocol adapter over the application invocation gateway.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | from contextlib import asynccontextmanager |
| 5 | from dataclasses import dataclass |
| 6 | from typing import Any |
| 7 | from uuid import UUID |
| 8 | |
| 9 | from mcp.server.mcpserver import Context, MCPServer |
| 10 | from mcp.types import ToolAnnotations |
| 11 | from pydantic import Field, ValidationError |
| 12 | from starlette.requests import Request |
| 13 | from starlette.responses import JSONResponse |
| 14 | |
| 15 | from app.application_client import ApplicationResearchReader, HttpApplicationResearchReader |
| 16 | from app.audit import McpAuditSink, StructuredMcpAuditLogger |
| 17 | from app.contracts import ( |
| 18 | McpAuthContext, |
| 19 | McpErrorCode, |
| 20 | McpGatewayError, |
| 21 | SAFE_ERROR_MESSAGES, |
| 22 | StrictContract, |
| 23 | normalize_request_id, |
| 24 | ) |
| 25 | from app.external import ( |
| 26 | ExternalMcpGateway, |
| 27 | McpServerRegistry, |
| 28 | ProviderFallbackAuthorization, |
| 29 | ) |
| 30 | from app.invocation import McpInvocationService |
| 31 | from app.policy import McpToolPolicy |
| 32 | from app.protocol import McpProtocolGuard |
| 33 | from app.registry import McpToolRegistry |
| 34 | from app.settings import McpGatewaySettings |
| 35 | from app.tools import build_internal_tool_registry |
| 36 | from app.yahoo_finance_mcp import ( |
| 37 | YahooFinanceCapabilityRegistry, |
| 38 | YahooFinanceMcpConfig, |
| 39 | YahooFinanceMcpProvider, |
| 40 | ) |
| 41 | |
| 42 | |
| 43 | class ExternalResearchAcquisitionRequest(StrictContract): |
| 44 | provider_id: str = Field(alias="providerId") |
| 45 | region: str |
| 46 | requirement_id: str = Field(alias="requirementId") |
| 47 | global_instrument_id: UUID = Field(alias="globalInstrumentId") |
| 48 | provider_symbol: str = Field(alias="providerSymbol", min_length=1, max_length=100) |
| 49 | expected_exchange: str | None = Field(default=None, alias="expectedExchange", max_length=100) |
| 50 | expected_currency: str | None = Field(default=None, alias="expectedCurrency", max_length=20) |
| 51 | authorization: ProviderFallbackAuthorization |
| 52 | |
| 53 | |
| 54 | @dataclass |
| 55 | class McpGatewayContainer: |
| 56 | settings: McpGatewaySettings |
| 57 | auth: McpAuthContext |
| 58 | reader: ApplicationResearchReader |
| 59 | registry: McpToolRegistry |
| 60 | audit: McpAuditSink |
| 61 | invocation: McpInvocationService |
| 62 | external_registry: McpServerRegistry |
| 63 | external_gateway: ExternalMcpGateway |
| 64 | |
| 65 | |
| 66 | def create_container( |
| 67 | settings: McpGatewaySettings, |
| 68 | *, |
| 69 | reader: ApplicationResearchReader | None = None, |
| 70 | audit: McpAuditSink | None = None, |
| 71 | ) -> McpGatewayContainer: |
| 72 | application_reader = reader or HttpApplicationResearchReader(settings) |
| 73 | registry = build_internal_tool_registry(application_reader) |
| 74 | audit_sink = audit or StructuredMcpAuditLogger(environment=settings.environment) |
| 75 | external_registry = McpServerRegistry() |
| 76 | if settings.yahoo_finance_mcp_enabled: |
| 77 | capabilities = YahooFinanceCapabilityRegistry.from_json( |
| 78 | settings.yahoo_finance_mcp_capabilities_json |
| 79 | ) |
| 80 | external_registry.register( |
| 81 | YahooFinanceMcpProvider( |
| 82 | YahooFinanceMcpConfig( |
| 83 | transport=settings.yahoo_finance_mcp_transport, |
| 84 | endpoint=settings.yahoo_finance_mcp_endpoint, |
| 85 | stdio_command=settings.yahoo_finance_mcp_stdio_command, |
| 86 | stdio_args=settings.yahoo_finance_mcp_stdio_args, |
| 87 | auth_type=settings.yahoo_finance_mcp_auth_type, |
| 88 | auth_header_name=settings.yahoo_finance_mcp_auth_header_name, |
| 89 | auth_token=settings.yahoo_finance_mcp_auth_token, |
| 90 | stdio_token_env_name=settings.yahoo_finance_mcp_stdio_token_env_name, |
| 91 | timeout_seconds=settings.yahoo_finance_mcp_timeout_seconds, |
| 92 | max_retries=settings.yahoo_finance_mcp_max_retries, |
| 93 | retry_backoff_seconds=settings.yahoo_finance_mcp_retry_backoff_seconds, |
| 94 | max_concurrency=settings.yahoo_finance_mcp_max_concurrency, |
| 95 | ), |
| 96 | capabilities, |
| 97 | ) |
| 98 | ) |
| 99 | return McpGatewayContainer( |
| 100 | settings=settings, |
| 101 | auth=settings.auth_context(), |
| 102 | reader=application_reader, |
| 103 | registry=registry, |
| 104 | audit=audit_sink, |
| 105 | invocation=McpInvocationService( |
| 106 | registry, |
| 107 | McpToolPolicy(), |
| 108 | audit_sink, |
| 109 | timeout_seconds=settings.invocation_timeout_seconds, |
| 110 | ), |
| 111 | external_registry=external_registry, |
| 112 | external_gateway=ExternalMcpGateway( |
| 113 | external_registry, |
| 114 | enabled=settings.external_providers_enabled, |
| 115 | timeout_seconds=settings.invocation_timeout_seconds, |
| 116 | audit=audit_sink, |
| 117 | auth=settings.auth_context(), |
| 118 | ), |
| 119 | ) |
| 120 | |
| 121 | |
| 122 | def create_mcp_server(container: McpGatewayContainer) -> MCPServer: |
| 123 | @asynccontextmanager |
| 124 | async def lifespan(_server): |
| 125 | try: |
| 126 | yield container |
| 127 | finally: |
| 128 | await container.reader.close() |
| 129 | |
| 130 | server = MCPServer( |
| 131 | "ai-investment-internal-mcp", |
| 132 | title="AI Investment Internal MCP", |
| 133 | description="Read-only internal application intelligence tools.", |
| 134 | instructions=( |
| 135 | "Use canonical globalInstrumentId values only. Tools read persisted application data " |
| 136 | "and never place trades, mutate portfolios, or trigger provider acquisition." |
| 137 | ), |
| 138 | version="0.2.0", |
| 139 | lifespan=lifespan, |
| 140 | warn_on_duplicate_tools=True, |
| 141 | middleware=[McpProtocolGuard(container.registry, container.audit, container.auth)], |
| 142 | ) |
| 143 | annotations = ToolAnnotations( |
| 144 | readOnlyHint=True, |
| 145 | destructiveHint=False, |
| 146 | idempotentHint=True, |
| 147 | openWorldHint=False, |
| 148 | ) |
| 149 | |
| 150 | @server.tool(annotations=annotations, structured_output=True) |
| 151 | async def get_research_readiness(globalInstrumentId: str, ctx: Context) -> dict[str, Any]: |
| 152 | """Read persisted Research Readiness for a canonical instrument; performs no acquisition.""" |
| 153 | return await _invoke( |
| 154 | container, |
| 155 | ctx, |
| 156 | "get_research_readiness", |
| 157 | {"globalInstrumentId": globalInstrumentId}, |
| 158 | ) |
| 159 | |
| 160 | @server.tool(annotations=annotations, structured_output=True) |
| 161 | async def get_company_analysis( |
| 162 | globalInstrumentId: str, ctx: Context, allowPartial: bool = False |
| 163 | ) -> dict[str, Any]: |
| 164 | """Return the existing STOCK_RULE_ENGINE_V1 result over persisted application data.""" |
| 165 | return await _invoke( |
| 166 | container, |
| 167 | ctx, |
| 168 | "get_company_analysis", |
| 169 | {"globalInstrumentId": globalInstrumentId, "allowPartial": allowPartial}, |
| 170 | ) |
| 171 | |
| 172 | @server.tool(annotations=annotations, structured_output=True) |
| 173 | async def get_financial_facts(globalInstrumentId: str, ctx: Context) -> dict[str, Any]: |
| 174 | """Read persisted financial statement facts for a canonical instrument.""" |
| 175 | return await _invoke( |
| 176 | container, |
| 177 | ctx, |
| 178 | "get_financial_facts", |
| 179 | {"globalInstrumentId": globalInstrumentId}, |
| 180 | ) |
| 181 | |
| 182 | @server.tool(annotations=annotations, structured_output=True) |
| 183 | async def get_quarterly_results(globalInstrumentId: str, ctx: Context) -> dict[str, Any]: |
| 184 | """Read persisted quarterly result facts for a canonical instrument.""" |
| 185 | return await _invoke( |
| 186 | container, |
| 187 | ctx, |
| 188 | "get_quarterly_results", |
| 189 | {"globalInstrumentId": globalInstrumentId}, |
| 190 | ) |
| 191 | |
| 192 | @server.tool(annotations=annotations, structured_output=True) |
| 193 | async def get_shareholding(globalInstrumentId: str, ctx: Context) -> dict[str, Any]: |
| 194 | """Read persisted shareholding snapshots for a canonical instrument.""" |
| 195 | return await _invoke( |
| 196 | container, |
| 197 | ctx, |
| 198 | "get_shareholding", |
| 199 | {"globalInstrumentId": globalInstrumentId}, |
| 200 | ) |
| 201 | |
| 202 | @server.tool(annotations=annotations, structured_output=True) |
| 203 | async def get_recent_news( |
| 204 | globalInstrumentId: str, ctx: Context, days: int = 30 |
| 205 | ) -> dict[str, Any]: |
| 206 | """Read persisted current news; days must be between 1 and 30 inclusive.""" |
| 207 | return await _invoke( |
| 208 | container, |
| 209 | ctx, |
| 210 | "get_recent_news", |
| 211 | {"globalInstrumentId": globalInstrumentId, "days": days}, |
| 212 | ) |
| 213 | |
| 214 | @server.tool(annotations=annotations, structured_output=True) |
| 215 | async def get_sector_performance( |
| 216 | region: str, sector: str, period: str, ctx: Context, limit: int = 5 |
| 217 | ) -> dict[str, Any]: |
| 218 | """Read durable Sector Performance for USA, EUROPE, or INDIA.""" |
| 219 | return await _invoke( |
| 220 | container, |
| 221 | ctx, |
| 222 | "get_sector_performance", |
| 223 | {"region": region, "sector": sector, "period": period, "limit": limit}, |
| 224 | ) |
| 225 | |
| 226 | @server.tool(annotations=annotations, structured_output=True) |
| 227 | async def search_research_evidence( |
| 228 | globalInstrumentId: str, query: str, ctx: Context, limit: int = 10 |
| 229 | ) -> dict[str, Any]: |
| 230 | """Search persisted research evidence metadata; never performs external search.""" |
| 231 | return await _invoke( |
| 232 | container, |
| 233 | ctx, |
| 234 | "search_research_evidence", |
| 235 | {"globalInstrumentId": globalInstrumentId, "query": query, "limit": limit}, |
| 236 | ) |
| 237 | |
| 238 | @server.tool(annotations=annotations, structured_output=True) |
| 239 | async def get_watchlist(watchlistId: str, ctx: Context) -> dict[str, Any]: |
| 240 | """Read the authenticated user's watchlist research projection.""" |
| 241 | return await _invoke(container, ctx, "get_watchlist", {"watchlistId": watchlistId}) |
| 242 | |
| 243 | @server.custom_route("/health", methods=["GET"], include_in_schema=False) |
| 244 | async def health(_request: Request) -> JSONResponse: |
| 245 | return JSONResponse( |
| 246 | { |
| 247 | "status": "ok", |
| 248 | "service": container.settings.service_name, |
| 249 | "registeredTools": len(container.registry.tools), |
| 250 | } |
| 251 | ) |
| 252 | |
| 253 | @server.custom_route("/health/ready", methods=["GET"], include_in_schema=False) |
| 254 | async def ready(_request: Request) -> JSONResponse: |
| 255 | # External providers are intentionally excluded from readiness in 5A. |
| 256 | return JSONResponse( |
| 257 | { |
| 258 | "status": "ready", |
| 259 | "service": container.settings.service_name, |
| 260 | "externalProviderDependency": False, |
| 261 | } |
| 262 | ) |
| 263 | |
| 264 | @server.custom_route( |
| 265 | "/internal/v1/external-research/acquire", methods=["POST"], include_in_schema=False |
| 266 | ) |
| 267 | async def acquire_external_research(request: Request) -> JSONResponse: |
| 268 | request_id = normalize_request_id( |
| 269 | request.headers.get("X-Request-ID") or request.headers.get("X-Correlation-ID") |
| 270 | ) |
| 271 | caller = str(request.headers.get("X-AIP-Service-Identity") or "").strip() |
| 272 | if caller not in container.settings.external_caller_identities: |
| 273 | return _external_error(McpErrorCode.UNAUTHORIZED, request_id, 401) |
| 274 | try: |
| 275 | command = ExternalResearchAcquisitionRequest.model_validate(await request.json()) |
| 276 | except (ValueError, ValidationError): |
| 277 | return _external_error(McpErrorCode.INVALID_ARGUMENT, request_id, 400) |
| 278 | try: |
| 279 | data = await container.external_gateway.invoke_requirement( |
| 280 | provider_id=command.provider_id, |
| 281 | region=command.region, |
| 282 | requirement_id=command.requirement_id, |
| 283 | global_instrument_id=command.global_instrument_id, |
| 284 | arguments={ |
| 285 | "providerSymbol": command.provider_symbol, |
| 286 | "expectedExchange": command.expected_exchange, |
| 287 | "expectedCurrency": command.expected_currency, |
| 288 | }, |
| 289 | request_id=request_id, |
| 290 | authorization=command.authorization, |
| 291 | ) |
| 292 | except McpGatewayError as exc: |
| 293 | status_code = 403 if exc.code in { |
| 294 | McpErrorCode.FORBIDDEN, |
| 295 | McpErrorCode.MCP_TOOL_DENIED, |
| 296 | } else 424 |
| 297 | return _external_error(exc.code, request_id, status_code) |
| 298 | return JSONResponse({"ok": True, "requestId": request_id, "data": data}) |
| 299 | |
| 300 | return server |
| 301 | |
| 302 | |
| 303 | def _external_error(code: McpErrorCode, request_id: str, status_code: int) -> JSONResponse: |
| 304 | return JSONResponse( |
| 305 | { |
| 306 | "ok": False, |
| 307 | "requestId": request_id, |
| 308 | "error": {"code": code.value, "message": SAFE_ERROR_MESSAGES[code]}, |
| 309 | }, |
| 310 | status_code=status_code, |
| 311 | ) |
| 312 | |
| 313 | |
| 314 | async def _invoke( |
| 315 | container: McpGatewayContainer, |
| 316 | context: Context, |
| 317 | tool: str, |
| 318 | arguments: dict[str, Any], |
| 319 | ) -> dict[str, Any]: |
| 320 | return await container.invocation.invoke( |
| 321 | tool, |
| 322 | arguments, |
| 323 | container.auth, |
| 324 | request_id=context.request_id, |
| 325 | ) |