| 1 | """CLI entry point for STDIO and stateless Streamable HTTP transports.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import argparse |
| 5 | import logging |
| 6 | import sys |
| 7 | |
| 8 | from mcp.server.transport_security import TransportSecuritySettings |
| 9 | |
| 10 | from app.server import create_container, create_mcp_server |
| 11 | from app.settings import McpGatewaySettings |
| 12 | |
| 13 | |
| 14 | def main() -> None: |
| 15 | settings = McpGatewaySettings() |
| 16 | parser = argparse.ArgumentParser(description="AI Investment internal MCP gateway") |
| 17 | parser.add_argument( |
| 18 | "--transport", |
| 19 | choices=("stdio", "streamable-http"), |
| 20 | default=settings.transport, |
| 21 | ) |
| 22 | args = parser.parse_args() |
| 23 | logging.basicConfig(level=logging.INFO, stream=sys.stderr, format="%(message)s") |
| 24 | if not settings.feature_enabled: |
| 25 | raise SystemExit("MCP gateway is disabled by AIP_FEATURE_MCP_ENABLED") |
| 26 | |
| 27 | server = create_mcp_server(create_container(settings)) |
| 28 | if args.transport == "stdio": |
| 29 | server.run("stdio") |
| 30 | return |
| 31 | server.run( |
| 32 | "streamable-http", |
| 33 | host=settings.host, |
| 34 | port=settings.port, |
| 35 | streamable_http_path=settings.streamable_http_path, |
| 36 | stateless_http=True, |
| 37 | json_response=True, |
| 38 | max_request_body_size=settings.max_request_body_bytes, |
| 39 | transport_security=TransportSecuritySettings( |
| 40 | enable_dns_rebinding_protection=True, |
| 41 | allowed_hosts=settings.allowed_hosts, |
| 42 | allowed_origins=settings.allowed_origins, |
| 43 | ), |
| 44 | ) |
| 45 | |
| 46 | |
| 47 | if __name__ == "__main__": |
| 48 | main() |