server.py•4.69 kB
import logging
import os
import signal
import sys
from importlib.metadata import version
import click
from fastmcp import FastMCP
# Server name
MCP_SERVER_NAME = "mcp-panther"
# Get log level from environment variable, default to WARNING if not set
log_level_name = os.environ.get("LOG_LEVEL", "WARNING")
# Convert string log level to logging constant
log_level = getattr(logging, log_level_name.upper(), logging.DEBUG)
# Configure logging
def configure_logging(log_file: str | None = None, *, force: bool = False) -> None:
"""Configure logging to stderr or the specified file.
This also reconfigures the ``FastMCP`` logger so that all FastMCP output
uses the same handler as the rest of the application.
"""
handler: logging.Handler
if log_file:
handler = logging.FileHandler(os.path.expanduser(log_file))
else:
handler = logging.StreamHandler(sys.stderr)
logging.basicConfig(
level=log_level,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[handler],
force=force,
)
# Ensure FastMCP logs propagate to the root logger
fastmcp_logger = logging.getLogger("FastMCP")
for hdlr in list(fastmcp_logger.handlers):
fastmcp_logger.removeHandler(hdlr)
fastmcp_logger.propagate = True
fastmcp_logger.setLevel(log_level)
configure_logging(os.environ.get("MCP_LOG_FILE"))
logger = logging.getLogger(MCP_SERVER_NAME)
# Support multiple import paths to accommodate different execution contexts:
# 1. When running as a binary, uvx expects relative imports
# 2. When running with MCP inspector: `uv run mcp dev src/mcp_panther/server.py`
# 3. When installing: `uv run mcp install src/mcp_panther/server.py`
try:
from panther_mcp_core.prompts.registry import register_all_prompts
from panther_mcp_core.resources.registry import register_all_resources
from panther_mcp_core.tools.registry import register_all_tools
except ImportError:
from .panther_mcp_core.prompts.registry import register_all_prompts
from .panther_mcp_core.resources.registry import register_all_resources
from .panther_mcp_core.tools.registry import register_all_tools
# Server dependencies
deps = [
"gql[aiohttp]",
"aiohttp",
]
# Create the MCP server
mcp = FastMCP(MCP_SERVER_NAME, dependencies=deps)
# Register all tools with MCP using the registry
register_all_tools(mcp)
# Register all prompts with MCP using the registry
register_all_prompts(mcp)
# Register all resources with MCP using the registry
register_all_resources(mcp)
def handle_signals():
def signal_handler(sig, frame):
logger.info(f"Received signal {sig}, shutting down...")
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
# SIGHUP is not available on Windows
if hasattr(signal, "SIGHUP"):
signal.signal(signal.SIGHUP, signal_handler)
@click.command()
@click.version_option(version("mcp-panther"), "--version", "-v")
@click.option(
"--transport",
type=click.Choice(["stdio", "streamable-http"]),
default=os.environ.get("MCP_TRANSPORT", default="stdio"),
help="Transport type (stdio or streamable-http)",
)
@click.option(
"--port",
default=int(os.environ.get("MCP_PORT", default="3000")),
help="Port to use for streamable HTTP transport",
)
@click.option(
"--host",
default=os.environ.get("MCP_HOST", default="127.0.0.1"),
help="Host to bind to for streamable HTTP transport",
)
@click.option(
"--log-file",
type=click.Path(),
default=os.environ.get("MCP_LOG_FILE"),
help="Write logs to this file instead of stderr",
)
def main(transport: str, port: int, host: str, log_file: str | None):
"""Run the Panther MCP server with the specified transport"""
# Set up signal handling
handle_signals()
# Reconfigure logging if a log file is provided
if log_file:
configure_logging(log_file, force=True)
major = sys.version_info.major
minor = sys.version_info.minor
micro = sys.version_info.micro
logger.info(f"Python {major}.{minor}.{micro}")
if transport == "streamable-http":
logger.info(
f"Starting Panther MCP Server with streamable HTTP transport on {host}:{port}"
)
try:
mcp.run(transport="streamable-http", host=host, port=port)
except KeyboardInterrupt:
logger.info("Keyboard interrupt received, forcing immediate exit")
sys.exit(0)
else:
logger.info("Starting Panther MCP Server with stdio transport")
# Let FastMCP handle all the asyncio details internally
mcp.run()