| 1 | """Central LOCAL/TEST/AZURE MCP gateway settings.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import json |
| 5 | import re |
| 6 | from typing import Literal |
| 7 | from uuid import UUID |
| 8 | |
| 9 | from pydantic import AliasChoices, Field, SecretStr, field_validator |
| 10 | from pydantic_settings import BaseSettings, SettingsConfigDict |
| 11 | |
| 12 | from app.contracts import McpAuthContext, McpAuthenticationType |
| 13 | |
| 14 | |
| 15 | class McpGatewaySettings(BaseSettings): |
| 16 | model_config = SettingsConfigDict(extra="ignore", case_sensitive=False) |
| 17 | |
| 18 | environment: Literal["LOCAL", "TEST", "AZURE"] = Field( |
| 19 | default="LOCAL", |
| 20 | validation_alias=AliasChoices("AIP_MCP_ENVIRONMENT", "AIP_ENVIRONMENT"), |
| 21 | ) |
| 22 | service_name: str = Field(default="mcp-gateway", validation_alias="AIP_MCP_SERVICE_NAME") |
| 23 | feature_enabled: bool = Field(default=True, validation_alias="AIP_FEATURE_MCP_ENABLED") |
| 24 | transport: Literal["stdio", "streamable-http"] = Field( |
| 25 | default="stdio", validation_alias="AIP_MCP_TRANSPORT" |
| 26 | ) |
| 27 | host: str = Field(default="127.0.0.1", validation_alias="AIP_MCP_HOST") |
| 28 | port: int = Field(default=8001, ge=1, le=65535, validation_alias="AIP_MCP_PORT") |
| 29 | streamable_http_path: str = Field(default="/mcp", validation_alias="AIP_MCP_HTTP_PATH") |
| 30 | max_request_body_bytes: int = Field( |
| 31 | default=262_144, ge=1024, le=4_194_304, validation_alias="AIP_MCP_MAX_REQUEST_BODY_BYTES" |
| 32 | ) |
| 33 | allowed_hosts_csv: str = Field( |
| 34 | default="localhost,localhost:*,127.0.0.1,127.0.0.1:*", |
| 35 | validation_alias="AIP_MCP_ALLOWED_HOSTS", |
| 36 | ) |
| 37 | allowed_origins_csv: str = Field(default="", validation_alias="AIP_MCP_ALLOWED_ORIGINS") |
| 38 | |
| 39 | research_base_url: str = Field( |
| 40 | default="http://127.0.0.1:8000", validation_alias="AIP_MCP_RESEARCH_BASE_URL" |
| 41 | ) |
| 42 | invocation_timeout_seconds: float = Field( |
| 43 | default=10.0, gt=0, le=120, validation_alias="AIP_MCP_INVOCATION_TIMEOUT_SECONDS" |
| 44 | ) |
| 45 | http_connect_timeout_seconds: float = Field( |
| 46 | default=3.0, gt=0, le=60, validation_alias="AIP_MCP_HTTP_CONNECT_TIMEOUT_SECONDS" |
| 47 | ) |
| 48 | http_read_timeout_seconds: float = Field( |
| 49 | default=8.0, gt=0, le=120, validation_alias="AIP_MCP_HTTP_READ_TIMEOUT_SECONDS" |
| 50 | ) |
| 51 | http_max_connections: int = Field( |
| 52 | default=50, ge=1, le=500, validation_alias="AIP_MCP_HTTP_MAX_CONNECTIONS" |
| 53 | ) |
| 54 | http_max_keepalive_connections: int = Field( |
| 55 | default=20, ge=0, le=200, validation_alias="AIP_MCP_HTTP_MAX_KEEPALIVE_CONNECTIONS" |
| 56 | ) |
| 57 | service_identity: str = Field( |
| 58 | default="local-mcp-gateway", validation_alias="AIP_MCP_SERVICE_IDENTITY" |
| 59 | ) |
| 60 | authentication_type: McpAuthenticationType | None = Field( |
| 61 | default=None, validation_alias="AIP_MCP_AUTHENTICATION_TYPE" |
| 62 | ) |
| 63 | local_user_id: UUID | None = Field(default=None, validation_alias="AIP_MCP_LOCAL_USER_ID") |
| 64 | roles_csv: str = Field(default="MCP_READER", validation_alias="AIP_MCP_ROLES") |
| 65 | scopes_csv: str = Field( |
| 66 | default="mcp:read", validation_alias="AIP_MCP_SCOPES" |
| 67 | ) |
| 68 | external_providers_enabled: bool = Field( |
| 69 | default=False, validation_alias="AIP_MCP_EXTERNAL_PROVIDERS_ENABLED" |
| 70 | ) |
| 71 | external_caller_identities_csv: str = Field( |
| 72 | default="research-engine", validation_alias="AIP_MCP_EXTERNAL_CALLER_IDENTITIES" |
| 73 | ) |
| 74 | yahoo_finance_mcp_enabled: bool = Field( |
| 75 | default=False, validation_alias="AIP_MCP_YAHOO_ENABLED" |
| 76 | ) |
| 77 | yahoo_finance_mcp_transport: Literal["streamable-http", "stdio"] = Field( |
| 78 | default="streamable-http", validation_alias="AIP_MCP_YAHOO_TRANSPORT" |
| 79 | ) |
| 80 | yahoo_finance_mcp_endpoint: str | None = Field( |
| 81 | default=None, validation_alias="AIP_MCP_YAHOO_ENDPOINT" |
| 82 | ) |
| 83 | yahoo_finance_mcp_stdio_command: str | None = Field( |
| 84 | default=None, validation_alias="AIP_MCP_YAHOO_STDIO_COMMAND" |
| 85 | ) |
| 86 | yahoo_finance_mcp_stdio_args_json: str = Field( |
| 87 | default="[]", validation_alias="AIP_MCP_YAHOO_STDIO_ARGS_JSON" |
| 88 | ) |
| 89 | yahoo_finance_mcp_auth_type: str = Field( |
| 90 | default="NONE", validation_alias="AIP_MCP_YAHOO_AUTH_TYPE" |
| 91 | ) |
| 92 | yahoo_finance_mcp_auth_header_name: str = Field( |
| 93 | default="Authorization", validation_alias="AIP_MCP_YAHOO_AUTH_HEADER_NAME" |
| 94 | ) |
| 95 | yahoo_finance_mcp_auth_token: SecretStr | None = Field( |
| 96 | default=None, validation_alias="AIP_MCP_YAHOO_AUTH_TOKEN" |
| 97 | ) |
| 98 | yahoo_finance_mcp_stdio_token_env_name: str | None = Field( |
| 99 | default=None, validation_alias="AIP_MCP_YAHOO_STDIO_TOKEN_ENV_NAME" |
| 100 | ) |
| 101 | yahoo_finance_mcp_capabilities_json: str = Field( |
| 102 | default="[]", validation_alias="AIP_MCP_YAHOO_CAPABILITIES_JSON" |
| 103 | ) |
| 104 | yahoo_finance_mcp_timeout_seconds: float = Field( |
| 105 | default=8.0, gt=0, le=120, validation_alias="AIP_MCP_YAHOO_TIMEOUT_SECONDS" |
| 106 | ) |
| 107 | yahoo_finance_mcp_max_retries: int = Field( |
| 108 | default=1, ge=0, le=3, validation_alias="AIP_MCP_YAHOO_MAX_RETRIES" |
| 109 | ) |
| 110 | yahoo_finance_mcp_retry_backoff_seconds: float = Field( |
| 111 | default=0.2, ge=0, le=10, validation_alias="AIP_MCP_YAHOO_RETRY_BACKOFF_SECONDS" |
| 112 | ) |
| 113 | yahoo_finance_mcp_max_concurrency: int = Field( |
| 114 | default=4, ge=1, le=32, validation_alias="AIP_MCP_YAHOO_MAX_CONCURRENCY" |
| 115 | ) |
| 116 | |
| 117 | @field_validator("streamable_http_path") |
| 118 | @classmethod |
| 119 | def valid_path(cls, value: str) -> str: |
| 120 | path = value.strip() |
| 121 | if not path.startswith("/"): |
| 122 | raise ValueError("AIP_MCP_HTTP_PATH must begin with /") |
| 123 | return path |
| 124 | |
| 125 | @field_validator("research_base_url") |
| 126 | @classmethod |
| 127 | def valid_research_url(cls, value: str) -> str: |
| 128 | url = value.strip().rstrip("/") |
| 129 | if not url.startswith(("http://", "https://")): |
| 130 | raise ValueError("AIP_MCP_RESEARCH_BASE_URL must be HTTP(S)") |
| 131 | return url |
| 132 | |
| 133 | @field_validator("yahoo_finance_mcp_endpoint") |
| 134 | @classmethod |
| 135 | def valid_yahoo_url(cls, value: str | None) -> str | None: |
| 136 | if value is None or not value.strip(): |
| 137 | return None |
| 138 | url = value.strip().rstrip("/") |
| 139 | if not url.startswith(("http://", "https://")): |
| 140 | raise ValueError("AIP_MCP_YAHOO_ENDPOINT must be HTTP(S)") |
| 141 | return url |
| 142 | |
| 143 | @field_validator("yahoo_finance_mcp_auth_type") |
| 144 | @classmethod |
| 145 | def valid_yahoo_auth_type(cls, value: str) -> str: |
| 146 | normalized = value.strip().upper() |
| 147 | if normalized not in {"NONE", "BEARER", "HEADER", "WORKLOAD_IDENTITY"}: |
| 148 | raise ValueError("Unsupported Yahoo MCP authentication type") |
| 149 | return normalized |
| 150 | |
| 151 | @field_validator("yahoo_finance_mcp_auth_header_name") |
| 152 | @classmethod |
| 153 | def valid_yahoo_auth_header(cls, value: str) -> str: |
| 154 | normalized = value.strip() |
| 155 | if not re.fullmatch(r"[A-Za-z0-9-]{1,100}", normalized): |
| 156 | raise ValueError("Yahoo MCP auth header name is invalid") |
| 157 | return normalized |
| 158 | |
| 159 | @field_validator("yahoo_finance_mcp_stdio_token_env_name") |
| 160 | @classmethod |
| 161 | def valid_yahoo_token_env(cls, value: str | None) -> str | None: |
| 162 | if value is None or not value.strip(): |
| 163 | return None |
| 164 | normalized = value.strip().upper() |
| 165 | if not re.fullmatch(r"[A-Z][A-Z0-9_]{1,99}", normalized): |
| 166 | raise ValueError("Yahoo MCP token environment name is invalid") |
| 167 | if normalized in {"PATH", "HOME", "SYSTEMROOT", "WINDIR", "TMP", "TEMP"}: |
| 168 | raise ValueError("Yahoo MCP token cannot replace a process environment setting") |
| 169 | return normalized |
| 170 | |
| 171 | @field_validator("yahoo_finance_mcp_stdio_args_json", "yahoo_finance_mcp_capabilities_json") |
| 172 | @classmethod |
| 173 | def valid_json_array(cls, value: str) -> str: |
| 174 | try: |
| 175 | parsed = json.loads(value) |
| 176 | except json.JSONDecodeError as exc: |
| 177 | raise ValueError("Yahoo MCP JSON configuration must be valid JSON") from exc |
| 178 | if not isinstance(parsed, list): |
| 179 | raise ValueError("Yahoo MCP JSON configuration must be an array") |
| 180 | return value |
| 181 | |
| 182 | @property |
| 183 | def allowed_hosts(self) -> list[str]: |
| 184 | return _csv(self.allowed_hosts_csv) |
| 185 | |
| 186 | @property |
| 187 | def allowed_origins(self) -> list[str]: |
| 188 | return _csv(self.allowed_origins_csv) |
| 189 | |
| 190 | @property |
| 191 | def external_caller_identities(self) -> frozenset[str]: |
| 192 | return frozenset(_csv(self.external_caller_identities_csv)) |
| 193 | |
| 194 | @property |
| 195 | def yahoo_finance_mcp_stdio_args(self) -> tuple[str, ...]: |
| 196 | return tuple(str(value) for value in json.loads(self.yahoo_finance_mcp_stdio_args_json)) |
| 197 | |
| 198 | def auth_context(self) -> McpAuthContext: |
| 199 | auth_type = self.authentication_type or ( |
| 200 | McpAuthenticationType.WORKLOAD_IDENTITY |
| 201 | if self.environment == "AZURE" |
| 202 | else McpAuthenticationType.LOCAL_SERVICE |
| 203 | ) |
| 204 | return McpAuthContext( |
| 205 | userId=self.local_user_id, |
| 206 | serviceIdentity=self.service_identity, |
| 207 | roles=tuple(_csv(self.roles_csv)), |
| 208 | scopes=tuple(_csv(self.scopes_csv)), |
| 209 | authenticationType=auth_type, |
| 210 | ) |
| 211 | |
| 212 | |
| 213 | def _csv(value: str) -> list[str]: |
| 214 | return [item.strip() for item in value.split(",") if item.strip()] |