| 1 | """Bounded client for existing provider-free application read contracts.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | from datetime import datetime, timedelta, timezone |
| 5 | from typing import Any, Protocol |
| 6 | |
| 7 | import httpx |
| 8 | from opentelemetry.propagate import inject |
| 9 | |
| 10 | from app.contracts import McpAuthContext, McpErrorCode, McpGatewayError |
| 11 | from app.settings import McpGatewaySettings |
| 12 | |
| 13 | |
| 14 | class ApplicationResearchReader(Protocol): |
| 15 | async def invoke( |
| 16 | self, |
| 17 | tool: str, |
| 18 | arguments: dict[str, Any], |
| 19 | auth: McpAuthContext, |
| 20 | request_id: str, |
| 21 | ) -> dict[str, Any]: ... |
| 22 | |
| 23 | async def close(self) -> None: ... |
| 24 | |
| 25 | |
| 26 | class HttpApplicationResearchReader: |
| 27 | """Maps MCP reads onto existing research-engine APIs without acquisition routes.""" |
| 28 | |
| 29 | def __init__( |
| 30 | self, |
| 31 | settings: McpGatewaySettings, |
| 32 | *, |
| 33 | client: httpx.AsyncClient | None = None, |
| 34 | ) -> None: |
| 35 | self.settings = settings |
| 36 | self._owns_client = client is None |
| 37 | self.client = client or httpx.AsyncClient( |
| 38 | base_url=settings.research_base_url, |
| 39 | timeout=httpx.Timeout( |
| 40 | connect=settings.http_connect_timeout_seconds, |
| 41 | read=settings.http_read_timeout_seconds, |
| 42 | write=settings.http_read_timeout_seconds, |
| 43 | pool=settings.http_connect_timeout_seconds, |
| 44 | ), |
| 45 | limits=httpx.Limits( |
| 46 | max_connections=settings.http_max_connections, |
| 47 | max_keepalive_connections=settings.http_max_keepalive_connections, |
| 48 | ), |
| 49 | follow_redirects=False, |
| 50 | ) |
| 51 | |
| 52 | async def close(self) -> None: |
| 53 | if self._owns_client: |
| 54 | await self.client.aclose() |
| 55 | |
| 56 | async def invoke( |
| 57 | self, |
| 58 | tool: str, |
| 59 | arguments: dict[str, Any], |
| 60 | auth: McpAuthContext, |
| 61 | request_id: str, |
| 62 | ) -> dict[str, Any]: |
| 63 | instrument_id = arguments.get("globalInstrumentId") |
| 64 | if tool == "get_research_readiness": |
| 65 | return await self._json( |
| 66 | "GET", |
| 67 | f"/api/v1/research/readiness/{instrument_id}", |
| 68 | auth, |
| 69 | request_id, |
| 70 | not_found_code=McpErrorCode.READINESS_NOT_AVAILABLE, |
| 71 | ) |
| 72 | if tool == "get_company_analysis": |
| 73 | return await self._json( |
| 74 | "POST", |
| 75 | f"/api/v1/research/analysis/{instrument_id}", |
| 76 | auth, |
| 77 | request_id, |
| 78 | json={"allowPartial": bool(arguments.get("allowPartial", False))}, |
| 79 | not_found_code=McpErrorCode.ANALYSIS_NOT_AVAILABLE, |
| 80 | ) |
| 81 | if tool in {"get_financial_facts", "get_quarterly_results", "get_shareholding", "get_recent_news"}: |
| 82 | summary = await self._json( |
| 83 | "GET", f"/api/v1/research/companies/{instrument_id}/summary", auth, request_id |
| 84 | ) |
| 85 | return _summary_projection(tool, summary, arguments) |
| 86 | if tool == "search_research_evidence": |
| 87 | documents = await self._json( |
| 88 | "GET", f"/api/v1/research/companies/{instrument_id}/documents", auth, request_id |
| 89 | ) |
| 90 | return _evidence_search(documents, arguments) |
| 91 | if tool == "get_sector_performance": |
| 92 | return await self._json( |
| 93 | "GET", |
| 94 | "/api/v1/research/sector-performance", |
| 95 | auth, |
| 96 | request_id, |
| 97 | params={ |
| 98 | "region": arguments["region"], |
| 99 | "sector": arguments["sector"], |
| 100 | "period": arguments["period"], |
| 101 | "limit": arguments["limit"], |
| 102 | }, |
| 103 | ) |
| 104 | if tool == "get_watchlist": |
| 105 | return await self._json( |
| 106 | "GET", |
| 107 | f"/api/v1/research/watchlists/{arguments['watchlistId']}/research", |
| 108 | auth, |
| 109 | request_id, |
| 110 | not_found_code=McpErrorCode.DOWNSTREAM_UNAVAILABLE, |
| 111 | ) |
| 112 | raise McpGatewayError(McpErrorCode.MCP_TOOL_NOT_FOUND) |
| 113 | |
| 114 | async def _json( |
| 115 | self, |
| 116 | method: str, |
| 117 | path: str, |
| 118 | auth: McpAuthContext, |
| 119 | request_id: str, |
| 120 | not_found_code: McpErrorCode = McpErrorCode.COMPANY_NOT_RESOLVED, |
| 121 | **kwargs: Any, |
| 122 | ) -> Any: |
| 123 | headers = { |
| 124 | "X-Request-ID": request_id, |
| 125 | "X-Correlation-Id": request_id, |
| 126 | **auth.identity_headers(), |
| 127 | } |
| 128 | inject(headers) |
| 129 | try: |
| 130 | response = await self.client.request(method, path, headers=headers, **kwargs) |
| 131 | except httpx.TimeoutException as exc: |
| 132 | raise McpGatewayError(McpErrorCode.DOWNSTREAM_TIMEOUT) from exc |
| 133 | except httpx.RequestError as exc: |
| 134 | raise McpGatewayError(McpErrorCode.DOWNSTREAM_UNAVAILABLE) from exc |
| 135 | if response.status_code in {401}: |
| 136 | raise McpGatewayError(McpErrorCode.UNAUTHORIZED) |
| 137 | if response.status_code in {403}: |
| 138 | raise McpGatewayError(McpErrorCode.FORBIDDEN) |
| 139 | if response.status_code == 404: |
| 140 | try: |
| 141 | detail = str(response.json().get("detail", "")).strip().upper() |
| 142 | except (AttributeError, ValueError): |
| 143 | detail = "" |
| 144 | code = ( |
| 145 | McpErrorCode.COMPANY_NOT_RESOLVED |
| 146 | if detail == McpErrorCode.COMPANY_NOT_RESOLVED.value |
| 147 | else not_found_code |
| 148 | ) |
| 149 | raise McpGatewayError(code) |
| 150 | if response.status_code in {400, 422}: |
| 151 | raise McpGatewayError(McpErrorCode.INVALID_ARGUMENT) |
| 152 | if response.status_code >= 400: |
| 153 | raise McpGatewayError(McpErrorCode.DOWNSTREAM_UNAVAILABLE) |
| 154 | try: |
| 155 | return response.json() |
| 156 | except ValueError as exc: |
| 157 | raise McpGatewayError(McpErrorCode.DOWNSTREAM_UNAVAILABLE) from exc |
| 158 | |
| 159 | |
| 160 | def _summary_projection(tool: str, summary: dict[str, Any], arguments: dict[str, Any]) -> dict[str, Any]: |
| 161 | instrument_id = arguments["globalInstrumentId"] |
| 162 | if tool == "get_financial_facts": |
| 163 | return { |
| 164 | "globalInstrumentId": instrument_id, |
| 165 | "financialResultHistory": summary.get("financialResultHistory", []), |
| 166 | "balanceSheetHistory": summary.get("balanceSheetHistory", []), |
| 167 | "cashFlowHistory": summary.get("cashFlowHistory", []), |
| 168 | "dataFreshness": summary.get("dataFreshness"), |
| 169 | } |
| 170 | if tool == "get_quarterly_results": |
| 171 | return { |
| 172 | "globalInstrumentId": instrument_id, |
| 173 | "latestQuarterlyResult": summary.get("latestQuarterlyResult"), |
| 174 | "financialResultHistory": summary.get("financialResultHistory", []), |
| 175 | "dataFreshness": summary.get("dataFreshness"), |
| 176 | } |
| 177 | if tool == "get_shareholding": |
| 178 | return { |
| 179 | "globalInstrumentId": instrument_id, |
| 180 | "shareholdingSnapshots": summary.get("shareholdingSnapshots", []), |
| 181 | "shareholdingFreshness": summary.get("shareholdingFreshness", "UNAVAILABLE"), |
| 182 | } |
| 183 | if tool == "get_recent_news": |
| 184 | days = int(arguments.get("days", 30)) |
| 185 | now = datetime.now(timezone.utc) |
| 186 | cutoff = now - timedelta(days=days) |
| 187 | items: list[dict[str, Any]] = [] |
| 188 | for event in summary.get("recentEvents", []): |
| 189 | publication_value = event.get("publishedAt") or event.get("eventDate") |
| 190 | publication_date = _parse_datetime(publication_value) |
| 191 | if publication_date is None or not cutoff <= publication_date <= now: |
| 192 | continue |
| 193 | items.append( |
| 194 | { |
| 195 | "eventId": event.get("eventId"), |
| 196 | "eventType": event.get("eventType"), |
| 197 | "publicationDate": publication_date.isoformat(), |
| 198 | "dateBasis": "PUBLISHED_AT" if event.get("publishedAt") else "EVENT_DATE", |
| 199 | "title": event.get("title"), |
| 200 | "summary": event.get("summary"), |
| 201 | "impact": event.get("impact"), |
| 202 | "source": { |
| 203 | "url": event.get("sourceUrl"), |
| 204 | "type": event.get("sourceType"), |
| 205 | "classification": event.get("sourceClassification"), |
| 206 | "reliability": event.get("reliability"), |
| 207 | }, |
| 208 | } |
| 209 | ) |
| 210 | return {"globalInstrumentId": instrument_id, "days": days, "news": items} |
| 211 | raise McpGatewayError(McpErrorCode.MCP_TOOL_NOT_FOUND) |
| 212 | |
| 213 | |
| 214 | def _evidence_search(documents: Any, arguments: dict[str, Any]) -> dict[str, Any]: |
| 215 | values = documents if isinstance(documents, list) else [] |
| 216 | query = str(arguments["query"]).casefold() |
| 217 | limit = int(arguments.get("limit", 10)) |
| 218 | matches = [] |
| 219 | for document in values: |
| 220 | haystack = " ".join( |
| 221 | str(document.get(key) or "") |
| 222 | for key in ("title", "sourceName", "publisher", "documentType", "documentSubtype") |
| 223 | ).casefold() |
| 224 | if query not in haystack: |
| 225 | continue |
| 226 | matches.append( |
| 227 | { |
| 228 | "documentId": document.get("documentId"), |
| 229 | "title": document.get("title"), |
| 230 | "canonicalUrl": document.get("canonicalUrl"), |
| 231 | "sourceName": document.get("sourceName"), |
| 232 | "publisher": document.get("publisher"), |
| 233 | "sourceClassification": document.get("sourceClassification"), |
| 234 | "reliabilityLevel": document.get("reliabilityLevel"), |
| 235 | "documentType": document.get("documentType"), |
| 236 | "documentSubtype": document.get("documentSubtype"), |
| 237 | "publishedAt": document.get("publishedAt"), |
| 238 | "retrievedAt": document.get("retrievedAt"), |
| 239 | } |
| 240 | ) |
| 241 | matches.sort(key=lambda item: item.get("publishedAt") or item.get("retrievedAt") or "", reverse=True) |
| 242 | return { |
| 243 | "globalInstrumentId": arguments["globalInstrumentId"], |
| 244 | "query": arguments["query"], |
| 245 | "matches": matches[:limit], |
| 246 | } |
| 247 | |
| 248 | |
| 249 | def _parse_datetime(value: Any) -> datetime | None: |
| 250 | if not value: |
| 251 | return None |
| 252 | try: |
| 253 | parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) |
| 254 | except ValueError: |
| 255 | return None |
| 256 | if parsed.tzinfo is None: |
| 257 | parsed = parsed.replace(tzinfo=timezone.utc) |
| 258 | return parsed.astimezone(timezone.utc) |