Spaces:
Running
Running
File size: 5,351 Bytes
a506ee1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """FastMCP run command implementation."""
import importlib.util
import re
import sys
from pathlib import Path
from typing import Any, Literal
from fastmcp.utilities.logging import get_logger
logger = get_logger("cli.run")
TransportType = Literal["stdio", "streamable-http", "sse"]
def is_url(path: str) -> bool:
"""Check if a string is a URL."""
url_pattern = re.compile(r"^https?://")
return bool(url_pattern.match(path))
def parse_file_path(server_spec: str) -> tuple[Path, str | None]:
"""Parse a file path that may include a server object specification.
Args:
server_spec: Path to file, optionally with :object suffix
Returns:
Tuple of (file_path, server_object)
"""
# First check if we have a Windows path (e.g., C:\...)
has_windows_drive = len(server_spec) > 1 and server_spec[1] == ":"
# Split on the last colon, but only if it's not part of the Windows drive letter
# and there's actually another colon in the string after the drive letter
if ":" in (server_spec[2:] if has_windows_drive else server_spec):
file_str, server_object = server_spec.rsplit(":", 1)
else:
file_str, server_object = server_spec, None
# Resolve the file path
file_path = Path(file_str).expanduser().resolve()
if not file_path.exists():
logger.error(f"File not found: {file_path}")
sys.exit(1)
if not file_path.is_file():
logger.error(f"Not a file: {file_path}")
sys.exit(1)
return file_path, server_object
def import_server(file: Path, server_object: str | None = None) -> Any:
"""Import a MCP server from a file.
Args:
file: Path to the file
server_object: Optional object name in format "module:object" or just "object"
Returns:
The server object
"""
# Add parent directory to Python path so imports can be resolved
file_dir = str(file.parent)
if file_dir not in sys.path:
sys.path.insert(0, file_dir)
# Import the module
spec = importlib.util.spec_from_file_location("server_module", file)
if not spec or not spec.loader:
logger.error("Could not load module", extra={"file": str(file)})
sys.exit(1)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# If no object specified, try common server names
if not server_object:
# Look for the most common server object names
for name in ["mcp", "server", "app"]:
if hasattr(module, name):
return getattr(module, name)
logger.error(
f"No server object found in {file}. Please either:\n"
"1. Use a standard variable name (mcp, server, or app)\n"
"2. Specify the object name with file:object syntax",
extra={"file": str(file)},
)
sys.exit(1)
# Handle module:object syntax
if ":" in server_object:
module_name, object_name = server_object.split(":", 1)
try:
server_module = importlib.import_module(module_name)
server = getattr(server_module, object_name, None)
except ImportError:
logger.error(
f"Could not import module '{module_name}'",
extra={"file": str(file)},
)
sys.exit(1)
else:
# Just object name
server = getattr(module, server_object, None)
if server is None:
logger.error(
f"Server object '{server_object}' not found",
extra={"file": str(file)},
)
sys.exit(1)
return server
def create_client_server(url: str) -> Any:
"""Create a FastMCP server from a client URL.
Args:
url: The URL to connect to
Returns:
A FastMCP server instance
"""
try:
import fastmcp
client = fastmcp.Client(url)
server = fastmcp.FastMCP.from_client(client)
return server
except Exception as e:
logger.error(f"Failed to create client for URL {url}: {e}")
sys.exit(1)
def run_command(
server_spec: str,
transport: str | None = None,
host: str | None = None,
port: int | None = None,
log_level: str | None = None,
) -> None:
"""Run a MCP server or connect to a remote one.
Args:
server_spec: Python file, object specification (file:obj), or URL
transport: Transport protocol to use
host: Host to bind to when using http transport
port: Port to bind to when using http transport
log_level: Log level
"""
if is_url(server_spec):
# Handle URL case
server = create_client_server(server_spec)
logger.debug(f"Created client proxy server for {server_spec}")
else:
# Handle file case
file, server_object = parse_file_path(server_spec)
server = import_server(file, server_object)
logger.debug(f'Found server "{server.name}" in {file}')
# Run the server
kwargs = {}
if transport:
kwargs["transport"] = transport
if host:
kwargs["host"] = host
if port:
kwargs["port"] = port
if log_level:
kwargs["log_level"] = log_level
try:
server.run(**kwargs)
except Exception as e:
logger.error(f"Failed to run server: {e}")
sys.exit(1)
|