|
| 1 | +import json |
| 2 | +import os |
1 | 3 | from typing import Annotated, Tuple |
2 | 4 | from urllib.parse import urlparse, urlunparse |
3 | 5 |
|
|
24 | 26 | DEFAULT_USER_AGENT_MANUAL = "ModelContextProtocol/1.0 (User-Specified; +https://github.com/modelcontextprotocol/servers)" |
25 | 27 |
|
26 | 28 |
|
| 29 | +def validate_declaration_manifest( |
| 30 | + manifest_raw: str | None, |
| 31 | + *, |
| 32 | + known_tools: set[str], |
| 33 | + known_resources: set[str], |
| 34 | + known_prompts: set[str], |
| 35 | +) -> None: |
| 36 | + """Validate declaration manifest and fail closed on unknown entries.""" |
| 37 | + if not manifest_raw: |
| 38 | + return |
| 39 | + |
| 40 | + try: |
| 41 | + manifest = json.loads(manifest_raw) |
| 42 | + except json.JSONDecodeError as exc: |
| 43 | + raise ValueError(f"Invalid declaration manifest JSON: {exc}") from exc |
| 44 | + |
| 45 | + if not isinstance(manifest, dict): |
| 46 | + raise ValueError("Declaration manifest must be a JSON object.") |
| 47 | + |
| 48 | + sections = { |
| 49 | + "tools": known_tools, |
| 50 | + "resources": known_resources, |
| 51 | + "prompts": known_prompts, |
| 52 | + } |
| 53 | + errors: list[str] = [] |
| 54 | + |
| 55 | + for key in manifest: |
| 56 | + if key not in sections: |
| 57 | + errors.append(f"manifest.{key}: unknown declaration section") |
| 58 | + |
| 59 | + for section, known in sections.items(): |
| 60 | + if section not in manifest: |
| 61 | + continue |
| 62 | + values = manifest[section] |
| 63 | + if not isinstance(values, list): |
| 64 | + errors.append(f"manifest.{section}: expected an array of strings") |
| 65 | + continue |
| 66 | + |
| 67 | + for index, entry in enumerate(values): |
| 68 | + if not isinstance(entry, str): |
| 69 | + errors.append( |
| 70 | + f"manifest.{section}[{index}]: expected string declaration name" |
| 71 | + ) |
| 72 | + continue |
| 73 | + if entry not in known: |
| 74 | + errors.append( |
| 75 | + f"manifest.{section}[{index}]: unknown declaration '{entry}'" |
| 76 | + ) |
| 77 | + |
| 78 | + if errors: |
| 79 | + raise ValueError( |
| 80 | + "Declaration manifest validation failed:\n" + "\n".join(errors) |
| 81 | + ) |
| 82 | + |
| 83 | + |
27 | 84 | def extract_content_from_html(html: str) -> str: |
28 | 85 | """Extract and convert HTML content to Markdown format. |
29 | 86 |
|
@@ -190,6 +247,13 @@ async def serve( |
190 | 247 | ignore_robots_txt: Whether to ignore robots.txt restrictions |
191 | 248 | proxy_url: Optional proxy URL to use for requests |
192 | 249 | """ |
| 250 | + validate_declaration_manifest( |
| 251 | + os.getenv("MCP_DECLARATION_MANIFEST"), |
| 252 | + known_tools={"fetch"}, |
| 253 | + known_resources=set(), |
| 254 | + known_prompts={"fetch"}, |
| 255 | + ) |
| 256 | + |
193 | 257 | server = Server("mcp-fetch") |
194 | 258 | user_agent_autonomous = custom_user_agent or DEFAULT_USER_AGENT_AUTONOMOUS |
195 | 259 | user_agent_manual = custom_user_agent or DEFAULT_USER_AGENT_MANUAL |
|
0 commit comments