| 1 | from __future__ import annotations |
| 2 | |
| 3 | import re |
| 4 | from typing import Literal |
| 5 | |
| 6 | from pydantic import AliasChoices, Field, field_validator |
| 7 | from pydantic_settings import BaseSettings, SettingsConfigDict |
| 8 | |
| 9 | |
| 10 | class YahooMcpSettings(BaseSettings): |
| 11 | model_config = SettingsConfigDict(extra="ignore", case_sensitive=False) |
| 12 | |
| 13 | environment: Literal["LOCAL", "TEST", "AZURE"] = Field( |
| 14 | default="LOCAL", |
| 15 | validation_alias=AliasChoices("AIP_YAHOO_MCP_ENVIRONMENT", "AIP_ENVIRONMENT"), |
| 16 | ) |
| 17 | service_name: str = Field( |
| 18 | default="yahoo-finance-mcp", validation_alias="AIP_YAHOO_MCP_SERVICE_NAME" |
| 19 | ) |
| 20 | transport: Literal["stdio", "streamable-http"] = Field( |
| 21 | default="stdio", validation_alias="AIP_YAHOO_MCP_TRANSPORT" |
| 22 | ) |
| 23 | host: str = Field(default="127.0.0.1", validation_alias="AIP_YAHOO_MCP_HOST") |
| 24 | port: int = Field(default=8002, ge=1, le=65535, validation_alias="AIP_YAHOO_MCP_PORT") |
| 25 | http_path: str = Field(default="/mcp", validation_alias="AIP_YAHOO_MCP_HTTP_PATH") |
| 26 | max_request_body_bytes: int = Field( |
| 27 | default=131_072, |
| 28 | ge=1024, |
| 29 | le=1_048_576, |
| 30 | validation_alias="AIP_YAHOO_MCP_MAX_REQUEST_BODY_BYTES", |
| 31 | ) |
| 32 | max_response_items: int = Field( |
| 33 | default=1000, ge=1, le=5000, validation_alias="AIP_YAHOO_MCP_MAX_RESPONSE_ITEMS" |
| 34 | ) |
| 35 | upstream_timeout_seconds: float = Field( |
| 36 | default=8.0, gt=0, le=60, validation_alias="AIP_YAHOO_MCP_UPSTREAM_TIMEOUT_SECONDS" |
| 37 | ) |
| 38 | max_concurrency: int = Field( |
| 39 | default=4, ge=1, le=32, validation_alias="AIP_YAHOO_MCP_MAX_CONCURRENCY" |
| 40 | ) |
| 41 | user_agent: str = Field( |
| 42 | default="AIInvestmentResearchBot/0.1 contact=research-compliance@example.invalid", |
| 43 | validation_alias="AIP_YAHOO_MCP_USER_AGENT", |
| 44 | ) |
| 45 | allowed_hosts_csv: str = Field( |
| 46 | default="localhost,localhost:*,127.0.0.1,127.0.0.1:*", |
| 47 | validation_alias="AIP_YAHOO_MCP_ALLOWED_HOSTS", |
| 48 | ) |
| 49 | allowed_origins_csv: str = Field( |
| 50 | default="", validation_alias="AIP_YAHOO_MCP_ALLOWED_ORIGINS" |
| 51 | ) |
| 52 | |
| 53 | @field_validator("http_path") |
| 54 | @classmethod |
| 55 | def valid_path(cls, value: str) -> str: |
| 56 | result = value.strip() |
| 57 | if not result.startswith("/"): |
| 58 | raise ValueError("AIP_YAHOO_MCP_HTTP_PATH must begin with /") |
| 59 | return result |
| 60 | |
| 61 | @field_validator("user_agent") |
| 62 | @classmethod |
| 63 | def bounded_user_agent(cls, value: str) -> str: |
| 64 | result = re.sub(r"[\r\n]", "", value).strip() |
| 65 | if not 1 <= len(result) <= 300: |
| 66 | raise ValueError("AIP_YAHOO_MCP_USER_AGENT is invalid") |
| 67 | return result |
| 68 | |
| 69 | @property |
| 70 | def allowed_hosts(self) -> list[str]: |
| 71 | return _csv(self.allowed_hosts_csv) |
| 72 | |
| 73 | @property |
| 74 | def allowed_origins(self) -> list[str]: |
| 75 | return _csv(self.allowed_origins_csv) |
| 76 | |
| 77 | |
| 78 | def _csv(value: str) -> list[str]: |
| 79 | return [item.strip() for item in value.split(",") if item.strip()] |