main
py 538 lines 19.2 KB
Raw
1 from __future__ import annotations
2
3 import os
4 import re
5 from contextlib import asynccontextmanager
6 from datetime import datetime, timedelta, timezone
7 from time import monotonic
8 from typing import Any, Awaitable, Callable
9 from uuid import UUID, uuid4
10
11 from mcp.server.context import HandlerResult, ServerRequestContext
12 from mcp.server.mcpserver import Context, MCPServer
13 from mcp.server.mcpserver.exceptions import ToolError
14 from mcp.types import CallToolResult, TextContent, ToolAnnotations
15 from opentelemetry import trace
16 from opentelemetry.propagate import extract
17 from pydantic import ValidationError
18 from starlette.requests import Request
19 from starlette.responses import JSONResponse
20
21 from yahoo_mcp_server.acquisition import YahooAcquisitionService, YahooMcpServiceError
22 from yahoo_mcp_server.audit import YahooMcpAudit
23 from yahoo_mcp_server.contracts import (
24 HistoryInput,
25 IdentityInput,
26 NewsInput,
27 RawArticle,
28 RawFact,
29 RawPrice,
30 RawProfile,
31 Region,
32 ToolPayload,
33 )
34 from yahoo_mcp_server.settings import YahooMcpSettings
35
36
37 REGISTERED_TOOLS = (
38 "get_quote",
39 "get_price_history",
40 "get_company_profile",
41 "get_financials",
42 "get_quarterly_financials",
43 "get_news",
44 "get_sector_industry",
45 "get_analyst_data",
46 )
47 _REQUEST_ID = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
48 _TRACER = trace.get_tracer("ai-investment.yahoo-finance-mcp")
49 _STRUCTURED_FINANCIAL_METRICS = frozenset(
50 {
51 "trailingEps",
52 "forwardEps",
53 "trailingPE",
54 "forwardPE",
55 "priceToBook",
56 "evToEbitda",
57 "marketCap",
58 "freeCashFlow",
59 "operatingCashFlow",
60 "roe",
61 "roa",
62 "roce",
63 "profitMargin",
64 "operatingMargin",
65 "revenueGrowth",
66 "earningsGrowth",
67 "totalCash",
68 "totalDebt",
69 "debtToEquity",
70 "currentRatio",
71 }
72 )
73 _ANALYST_METRICS = frozenset(
74 {
75 "publicAnalystTargetLowPrice",
76 "publicAnalystTargetMedianPrice",
77 "publicAnalystTargetMeanPrice",
78 "publicAnalystTargetHighPrice",
79 "publicAnalystCount",
80 "publicAnalystRecommendationMean",
81 "publicAnalystConsensus",
82 }
83 )
84 _BASE_ARGUMENTS = {
85 "globalInstrumentId",
86 "verifiedYahooSymbol",
87 "region",
88 "exchange",
89 "currency",
90 }
91 _TOOL_ARGUMENTS = {
92 **{name: _BASE_ARGUMENTS for name in REGISTERED_TOOLS},
93 "get_price_history": _BASE_ARGUMENTS | {"lookbackDays"},
94 "get_news": _BASE_ARGUMENTS | {"days"},
95 }
96
97
98 class StrictToolArguments:
99 """Reject unknown tools and undeclared fields before SDK argument binding."""
100
101 async def __call__(self, context: ServerRequestContext, call_next) -> HandlerResult:
102 if context.method != "tools/call":
103 return await call_next(context)
104 params = context.params or {}
105 tool = str(params.get("name") or "")
106 allowed = _TOOL_ARGUMENTS.get(tool)
107 arguments = params.get("arguments") or {}
108 if allowed is None or not isinstance(arguments, dict) or set(arguments) - allowed:
109 return CallToolResult(
110 content=[TextContent(type="text", text="YAHOO_MCP_INVALID_ARGUMENT")],
111 isError=True,
112 )
113 return await call_next(context)
114
115
116 def create_yahoo_mcp_server(
117 settings: YahooMcpSettings | None = None,
118 *,
119 acquisition: YahooAcquisitionService | None = None,
120 audit: YahooMcpAudit | None = None,
121 ) -> MCPServer:
122 config = settings or YahooMcpSettings()
123 source = acquisition or YahooAcquisitionService(config)
124 audit_sink = audit or YahooMcpAudit(
125 service=config.service_name, environment=config.environment
126 )
127 @asynccontextmanager
128 async def lifespan(_server):
129 try:
130 yield source
131 finally:
132 await source.close()
133
134 server = MCPServer(
135 "ai-investment-yahoo-finance-mcp",
136 title="AI Investment First-party Yahoo Finance MCP",
137 description="Controlled read-only Yahoo Finance acquisition using verified mappings.",
138 instructions=(
139 "Accept only canonical globalInstrumentId values and application-verified Yahoo symbols. "
140 "This server never discovers or persists identity and exposes no write capability."
141 ),
142 version="0.1.0",
143 lifespan=lifespan,
144 warn_on_duplicate_tools=True,
145 middleware=[StrictToolArguments()],
146 )
147 annotations = ToolAnnotations(
148 readOnlyHint=True,
149 destructiveHint=False,
150 idempotentHint=True,
151 openWorldHint=True,
152 )
153
154 @server.tool(annotations=annotations, structured_output=True)
155 async def get_quote(
156 globalInstrumentId: UUID,
157 verifiedYahooSymbol: str,
158 region: Region,
159 ctx: Context,
160 exchange: str | None = None,
161 currency: str | None = None,
162 ) -> dict[str, Any]:
163 """Return a current Yahoo quote for an application-verified provider mapping."""
164 identity = _identity(globalInstrumentId, verifiedYahooSymbol, region, exchange, currency)
165
166 async def work() -> dict[str, Any]:
167 snapshot = await source.snapshot(identity)
168 fact = snapshot.facts.get("latestPrice")
169 if fact is None:
170 raise YahooMcpServiceError("YAHOO_MCP_INCOMPLETE")
171 return _payload(identity, snapshot, price=fact.value).wire()
172
173 return await _execute("get_quote", identity, ctx, audit_sink, work)
174
175 @server.tool(annotations=annotations, structured_output=True)
176 async def get_price_history(
177 globalInstrumentId: UUID,
178 verifiedYahooSymbol: str,
179 region: Region,
180 ctx: Context,
181 exchange: str | None = None,
182 currency: str | None = None,
183 lookbackDays: int = 400,
184 ) -> dict[str, Any]:
185 """Return bounded historical close observations for a verified Yahoo symbol."""
186 command = HistoryInput(
187 globalInstrumentId=globalInstrumentId,
188 verifiedYahooSymbol=verifiedYahooSymbol,
189 region=region,
190 exchange=exchange,
191 currency=currency,
192 lookbackDays=lookbackDays,
193 )
194
195 async def work() -> dict[str, Any]:
196 observations = await source.closes(command, lookback_days=command.lookback_days)
197 return ToolPayload(
198 globalInstrumentId=command.global_instrument_id,
199 symbol=command.verified_yahoo_symbol,
200 exchange=command.exchange,
201 currency=command.currency,
202 asOf=observations[-1].observed_at,
203 retrievedAt=datetime.now(timezone.utc),
204 sourceUrl=observations[-1].source_url,
205 prices=tuple(
206 RawPrice(
207 observedAt=item.observed_at,
208 close=item.price,
209 currency=item.currency or command.currency,
210 )
211 for item in observations
212 ),
213 ).wire()
214
215 return await _execute("get_price_history", command, ctx, audit_sink, work)
216
217 @server.tool(annotations=annotations, structured_output=True)
218 async def get_company_profile(
219 globalInstrumentId: UUID,
220 verifiedYahooSymbol: str,
221 region: Region,
222 ctx: Context,
223 exchange: str | None = None,
224 currency: str | None = None,
225 ) -> dict[str, Any]:
226 """Return the normalized Yahoo company profile without identity discovery."""
227 identity = _identity(globalInstrumentId, verifiedYahooSymbol, region, exchange, currency)
228
229 async def work() -> dict[str, Any]:
230 snapshot = await source.snapshot(identity)
231 profile = _profile(snapshot)
232 if not any((profile.company_name, profile.sector, profile.industry)):
233 raise YahooMcpServiceError("YAHOO_MCP_INCOMPLETE")
234 return _payload(identity, snapshot, profile=profile).wire()
235
236 return await _execute("get_company_profile", identity, ctx, audit_sink, work)
237
238 @server.tool(annotations=annotations, structured_output=True)
239 async def get_financials(
240 globalInstrumentId: UUID,
241 verifiedYahooSymbol: str,
242 region: Region,
243 ctx: Context,
244 exchange: str | None = None,
245 currency: str | None = None,
246 ) -> dict[str, Any]:
247 """Return normalized annual statements and supported valuation inputs."""
248 identity = _identity(globalInstrumentId, verifiedYahooSymbol, region, exchange, currency)
249
250 async def work() -> dict[str, Any]:
251 snapshot = await source.snapshot(identity)
252 facts = _statement_facts(snapshot, "ANNUAL") + _selected_facts(
253 snapshot, _STRUCTURED_FINANCIAL_METRICS
254 )
255 facts = facts[: config.max_response_items]
256 if not facts:
257 raise YahooMcpServiceError("YAHOO_MCP_INCOMPLETE")
258 latest = snapshot.facts.get("latestPrice")
259 return _payload(
260 identity,
261 snapshot,
262 price=latest.value if latest else None,
263 facts=tuple(facts),
264 ).wire()
265
266 return await _execute("get_financials", identity, ctx, audit_sink, work)
267
268 @server.tool(annotations=annotations, structured_output=True)
269 async def get_quarterly_financials(
270 globalInstrumentId: UUID,
271 verifiedYahooSymbol: str,
272 region: Region,
273 ctx: Context,
274 exchange: str | None = None,
275 currency: str | None = None,
276 ) -> dict[str, Any]:
277 """Return normalized quarterly statement facts with explicit period identity."""
278 identity = _identity(globalInstrumentId, verifiedYahooSymbol, region, exchange, currency)
279
280 async def work() -> dict[str, Any]:
281 snapshot = await source.snapshot(identity)
282 facts = _statement_facts(snapshot, "QUARTERLY")
283 facts = facts[: config.max_response_items]
284 if not facts:
285 raise YahooMcpServiceError("YAHOO_MCP_INCOMPLETE")
286 return _payload(identity, snapshot, facts=tuple(facts)).wire()
287
288 return await _execute("get_quarterly_financials", identity, ctx, audit_sink, work)
289
290 @server.tool(annotations=annotations, structured_output=True)
291 async def get_news(
292 globalInstrumentId: UUID,
293 verifiedYahooSymbol: str,
294 region: Region,
295 ctx: Context,
296 exchange: str | None = None,
297 currency: str | None = None,
298 days: int = 30,
299 ) -> dict[str, Any]:
300 """Return issuer-scoped Yahoo news published during the last 30 days at most."""
301 command = NewsInput(
302 globalInstrumentId=globalInstrumentId,
303 verifiedYahooSymbol=verifiedYahooSymbol,
304 region=region,
305 exchange=exchange,
306 currency=currency,
307 days=days,
308 )
309
310 async def work() -> dict[str, Any]:
311 snapshot = await source.snapshot(command)
312 cutoff = datetime.now(timezone.utc) - timedelta(days=command.days)
313 articles = []
314 seen = set()
315 for item in snapshot.news:
316 published = item.get("publishedAt")
317 url = str(item.get("url") or "")
318 if not isinstance(published, datetime) or published < cutoff or url in seen:
319 continue
320 seen.add(url)
321 articles.append(
322 RawArticle(
323 headline=item.get("headline"),
324 url=url,
325 publishedAt=published,
326 publisher=item.get("publisher") or "Yahoo Finance",
327 issuerSymbol=command.verified_yahoo_symbol,
328 summary=item.get("summary"),
329 )
330 )
331 if len(articles) >= config.max_response_items:
332 break
333 payload = _payload(command, snapshot, news=tuple(articles)).wire()
334 payload["newsQuerySucceeded"] = True
335 return payload
336
337 return await _execute("get_news", command, ctx, audit_sink, work)
338
339 @server.tool(annotations=annotations, structured_output=True)
340 async def get_sector_industry(
341 globalInstrumentId: UUID,
342 verifiedYahooSymbol: str,
343 region: Region,
344 ctx: Context,
345 exchange: str | None = None,
346 currency: str | None = None,
347 ) -> dict[str, Any]:
348 """Return Yahoo sector and industry labels as supporting evidence."""
349 identity = _identity(globalInstrumentId, verifiedYahooSymbol, region, exchange, currency)
350
351 async def work() -> dict[str, Any]:
352 snapshot = await source.snapshot(identity)
353 profile = _profile(snapshot)
354 if not profile.sector and not profile.industry:
355 raise YahooMcpServiceError("YAHOO_MCP_INCOMPLETE")
356 return _payload(identity, snapshot, profile=profile).wire()
357
358 return await _execute("get_sector_industry", identity, ctx, audit_sink, work)
359
360 @server.tool(annotations=annotations, structured_output=True)
361 async def get_analyst_data(
362 globalInstrumentId: UUID,
363 verifiedYahooSymbol: str,
364 region: Region,
365 ctx: Context,
366 exchange: str | None = None,
367 currency: str | None = None,
368 ) -> dict[str, Any]:
369 """Return only Yahoo analyst fields that are present; never derive a recommendation."""
370 identity = _identity(globalInstrumentId, verifiedYahooSymbol, region, exchange, currency)
371
372 async def work() -> dict[str, Any]:
373 snapshot = await source.snapshot(identity)
374 facts = _selected_facts(snapshot, _ANALYST_METRICS)
375 if not facts:
376 raise YahooMcpServiceError("YAHOO_MCP_INCOMPLETE")
377 return _payload(identity, snapshot, facts=tuple(facts)).wire()
378
379 return await _execute("get_analyst_data", identity, ctx, audit_sink, work)
380
381 @server.custom_route("/health", methods=["GET"], include_in_schema=False)
382 async def health(_request: Request) -> JSONResponse:
383 return JSONResponse(
384 {"status": "ok", "service": config.service_name, "registeredTools": len(REGISTERED_TOOLS)}
385 )
386
387 @server.custom_route("/health/ready", methods=["GET"], include_in_schema=False)
388 async def ready(_request: Request) -> JSONResponse:
389 return JSONResponse(
390 {"status": "ready", "service": config.service_name, "upstreamRequired": False}
391 )
392
393 # MCP 2.2.0 derives function schemas but leaves object extras unspecified.
394 # The pinned SDK exposes registered Tool metadata through ToolManager; mark
395 # the advertised schema as strict to match the protocol-edge guard above.
396 for registered in server._tool_manager.list_tools():
397 registered.parameters["additionalProperties"] = False
398
399 return server
400
401
402 def _identity(
403 global_instrument_id: UUID,
404 symbol: str,
405 region: Region,
406 exchange: str | None,
407 currency: str | None,
408 ) -> IdentityInput:
409 return IdentityInput(
410 globalInstrumentId=global_instrument_id,
411 verifiedYahooSymbol=symbol,
412 region=region,
413 exchange=exchange,
414 currency=currency,
415 )
416
417
418 async def _execute(
419 tool: str,
420 identity: IdentityInput,
421 context: Context,
422 audit: YahooMcpAudit,
423 work: Callable[[], Awaitable[dict[str, Any]]],
424 ) -> dict[str, Any]:
425 started = monotonic()
426 request_id = _request_id(context)
427 common = {
428 "request_id": request_id,
429 "global_instrument_id": str(identity.global_instrument_id),
430 "symbol": identity.verified_yahoo_symbol,
431 "tool": tool,
432 "started": started,
433 }
434 parent_context = extract(dict(context.headers or {}))
435 with _TRACER.start_as_current_span(f"yahoo_mcp.{tool}", context=parent_context):
436 audit.emit("YAHOO_MCP_REQUEST", **common)
437 try:
438 result = await work()
439 except YahooMcpServiceError as exc:
440 event = (
441 "YAHOO_MCP_IDENTITY_MISMATCH"
442 if exc.code == "YAHOO_MCP_IDENTITY_MISMATCH"
443 else "YAHOO_MCP_UNSUPPORTED"
444 if exc.code == "YAHOO_MCP_UNSUPPORTED"
445 else "YAHOO_MCP_FAILED"
446 )
447 audit.emit(event, **common, error_code=exc.code)
448 raise ToolError(exc.code) from None
449 except (ValidationError, ValueError):
450 audit.emit("YAHOO_MCP_FAILED", **common, error_code="YAHOO_MCP_INVALID_RESPONSE")
451 raise ToolError("YAHOO_MCP_INVALID_RESPONSE") from None
452 except Exception:
453 audit.emit(
454 "YAHOO_MCP_FAILED",
455 **common,
456 error_code="YAHOO_MCP_UPSTREAM_UNAVAILABLE",
457 )
458 raise ToolError("YAHOO_MCP_UPSTREAM_UNAVAILABLE") from None
459 audit.emit("YAHOO_MCP_SUCCESS", **common)
460 return result
461
462
463 def _request_id(context: Context) -> str:
464 headers = context.headers or {}
465 candidate = str(
466 headers.get("x-request-id")
467 or headers.get("X-Request-ID")
468 or headers.get("x-correlation-id")
469 or os.environ.get("AIP_REQUEST_ID")
470 or context.request_id
471 or ""
472 )
473 return candidate if _REQUEST_ID.fullmatch(candidate) else str(uuid4())
474
475
476 def _payload(identity: IdentityInput, snapshot, **updates: Any) -> ToolPayload:
477 return ToolPayload(
478 globalInstrumentId=identity.global_instrument_id,
479 symbol=identity.verified_yahoo_symbol,
480 exchange=snapshot.resolution.exchange or identity.exchange,
481 currency=snapshot.resolution.currency or identity.currency,
482 asOf=snapshot.market_as_of or snapshot.retrieved_at,
483 retrievedAt=snapshot.retrieved_at,
484 sourceUrl=snapshot.source_url,
485 **updates,
486 )
487
488
489 def _profile(snapshot) -> RawProfile:
490 def text(metric: str) -> str | None:
491 fact = snapshot.facts.get(metric)
492 return str(fact.value) if fact and fact.value not in (None, "") else None
493
494 return RawProfile(
495 companyName=text("providerCompanyName"),
496 sector=text("sector"),
497 industry=text("industry"),
498 )
499
500
501 def _selected_facts(snapshot, metrics: frozenset[str]) -> list[RawFact]:
502 result = []
503 for metric in sorted(metrics):
504 value = snapshot.facts.get(metric)
505 if value is None:
506 continue
507 result.append(
508 RawFact(
509 metric=metric,
510 value=value.value,
511 unit=value.unit,
512 asOf=value.as_of_date,
513 publishedAt=value.published_at,
514 confidence=value.confidence or 0.78,
515 rawFieldOrigin=value.calculation_basis or metric,
516 )
517 )
518 return result
519
520
521 def _statement_facts(snapshot, period_type: str) -> list[RawFact]:
522 values = []
523 for item in snapshot.statement_facts:
524 if str(item.get("periodType") or "").upper() != period_type:
525 continue
526 values.append(
527 RawFact(
528 metric=item.get("metric"),
529 value=item.get("value"),
530 periodEnd=item.get("periodEnd"),
531 periodType=period_type,
532 reportingBasis=item.get("reportingBasis") or "UNKNOWN",
533 publishedAt=item.get("publishedAt"),
534 confidence=item.get("confidence") or 0.78,
535 rawFieldOrigin=item.get("rawFieldOrigin"),
536 )
537 )
538 return values