(
host: str = "127.0.0.1",
port: int = 8000,
api_key: Optional[str] = "",
cors_allow_origins=["*"],
**kwargs,
)
| 737 | |
| 738 | |
| 739 | async def run( |
| 740 | host: str = "127.0.0.1", |
| 741 | port: int = 8000, |
| 742 | api_key: Optional[str] = "", |
| 743 | cors_allow_origins=["*"], |
| 744 | **kwargs, |
| 745 | ): |
| 746 | hot_reload = kwargs.get("hot_reload", False) |
| 747 | # Server API Key |
| 748 | api_dependency = get_verify_api_key(api_key) if api_key else None |
| 749 | connection_timeout = kwargs.get("connection_timeout", None) |
| 750 | strict_auth = kwargs.get("strict_auth", False) |
| 751 | |
| 752 | # MCP Server |
| 753 | server_type = normalize_server_type(kwargs.get("server_type")) |
| 754 | server_command = kwargs.get("server_command") |
| 755 | |
| 756 | # MCP Config |
| 757 | config_path = kwargs.get("config_path") |
| 758 | |
| 759 | # mcpo server |
| 760 | name = kwargs.get("name") or "MCP OpenAPI Proxy" |
| 761 | description = ( |
| 762 | kwargs.get("description") or "Automatically generated API from MCP Tool Schemas" |
| 763 | ) |
| 764 | version = kwargs.get("version") or "1.0" |
| 765 | |
| 766 | ssl_certfile = kwargs.get("ssl_certfile") |
| 767 | ssl_keyfile = kwargs.get("ssl_keyfile") |
| 768 | path_prefix = kwargs.get("path_prefix") or "/" |
| 769 | root_path = kwargs.get("root_path") or "" |
| 770 | |
| 771 | # Configure logging based on LOG_LEVEL environment variable |
| 772 | log_level = os.getenv("LOG_LEVEL", "INFO").upper() |
| 773 | try: |
| 774 | numeric_level = getattr(logging, log_level, None) |
| 775 | if not isinstance(numeric_level, int): |
| 776 | raise ValueError(f"Invalid log level: {log_level}") |
| 777 | except (ValueError, AttributeError): |
| 778 | logger.warning(f"Invalid LOG_LEVEL '{log_level}', defaulting to INFO") |
| 779 | numeric_level = logging.INFO |
| 780 | |
| 781 | logging.basicConfig( |
| 782 | level=numeric_level, format="%(asctime)s - %(levelname)s - %(message)s" |
| 783 | ) |
| 784 | |
| 785 | # Suppress HTTP request logs |
| 786 | class HTTPRequestFilter(logging.Filter): |
| 787 | def filter(self, record): |
| 788 | return not ( |
| 789 | record.levelname == "INFO" and "HTTP Request:" in record.getMessage() |
| 790 | ) |
| 791 | |
| 792 | # Apply filter to suppress HTTP request logs |
| 793 | logging.getLogger("uvicorn.access").addFilter(HTTPRequestFilter()) |
| 794 | logging.getLogger("httpx.access").addFilter(HTTPRequestFilter()) |
| 795 | logger.info("Starting MCPO Server...") |
| 796 | logger.info(f" Name: {name}") |
no test coverage detected