aljaz/mcp-python-sdk
1
0
Fork 0
mirror of https://github.com/aljazceru/mcp-python-sdk.git synced 2025-12-19 06:54:18 +01:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add CLI package

Browse Source
This commit is contained in:
David Soria Parra
2024-12-18 15:26:02 +00:00
parent 15cd7df87f
commit 87cee0ff33
5 changed files with 646 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pyproject.toml
View File

@@ -1,6 +1,6 @@
[project]
name = "mcp"
version = "1.1.2.dev0"
version = "1.2.0.dev0"
description = "Model Context Protocol SDK"
readme = "README.md"
requires-python = ">=3.10"
@@ -33,6 +33,10 @@ dependencies = [
[project.optional-dependencies]
rich = ["rich>=13.9.4"]
cli = ["typer>=0.12.4", "python-dotenv>=1.0.0"]
[project.scripts]
mcp = "mcp.cli:app [cli]"
[tool.uv]
resolution = "lowest-direct"

7
src/mcp/cli/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
"""FastMCP CLI package."""
from .cli import app
if __name__ == "__main__":
app()

137
src/mcp/cli/claude.py Normal file
View File

@@ -0,0 +1,137 @@
"""Claude app integration utilities."""
import json
import sys
from pathlib import Path
from mcp.server.fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
def get_claude_config_path() -> Path | None:
"""Get the Claude config directory based on platform."""
if sys.platform == "win32":
path = Path(Path.home(), "AppData", "Roaming", "Claude")
elif sys.platform == "darwin":
path = Path(Path.home(), "Library", "Application Support", "Claude")
else:
return None
if path.exists():
return path
return None
def update_claude_config(
file_spec: str,
server_name: str,
*,
with_editable: Path | None = None,
with_packages: list[str] | None = None,
env_vars: dict[str, str] | None = None,
) -> bool:
"""Add or update a FastMCP server in Claude's configuration.
Args:
file_spec: Path to the server file, optionally with :object suffix
server_name: Name for the server in Claude's config
with_editable: Optional directory to install in editable mode
with_packages: Optional list of additional packages to install
env_vars: Optional dictionary of environment variables. These are merged with
any existing variables, with new values taking precedence.
Raises:
RuntimeError: If Claude Desktop's config directory is not found, indicating
Claude Desktop may not be installed or properly set up.
"""
config_dir = get_claude_config_path()
if not config_dir:
raise RuntimeError(
"Claude Desktop config directory not found. Please ensure Claude Desktop "
"is installed and has been run at least once to initialize its configuration."
)
config_file = config_dir / "claude_desktop_config.json"
if not config_file.exists():
try:
config_file.write_text("{}")
except Exception as e:
logger.error(
"Failed to create Claude config file",
extra={
"error": str(e),
"config_file": str(config_file),
},
)
return False
try:
config = json.loads(config_file.read_text())
if "mcpServers" not in config:
config["mcpServers"] = {}
# Always preserve existing env vars and merge with new ones
if (
server_name in config["mcpServers"]
and "env" in config["mcpServers"][server_name]
):
existing_env = config["mcpServers"][server_name]["env"]
if env_vars:
# New vars take precedence over existing ones
env_vars = {**existing_env, **env_vars}
else:
env_vars = existing_env
# Build uv run command
args = ["run"]
# Collect all packages in a set to deduplicate
packages = {"fastmcp"}
if with_packages:
packages.update(pkg for pkg in with_packages if pkg)
# Add all packages with --with
for pkg in sorted(packages):
args.extend(["--with", pkg])
if with_editable:
args.extend(["--with-editable", str(with_editable)])
# Convert file path to absolute before adding to command
# Split off any :object suffix first
if ":" in file_spec:
file_path, server_object = file_spec.rsplit(":", 1)
file_spec = f"{Path(file_path).resolve()}:{server_object}"
else:
file_spec = str(Path(file_spec).resolve())
# Add fastmcp run command
args.extend(["fastmcp", "run", file_spec])
server_config = {
"command": "uv",
"args": args,
}
# Add environment variables if specified
if env_vars:
server_config["env"] = env_vars
config["mcpServers"][server_name] = server_config
config_file.write_text(json.dumps(config, indent=2))
logger.info(
f"Added server '{server_name}' to Claude config",
extra={"config_file": str(config_file)},
)
return True
except Exception as e:
logger.error(
"Failed to update Claude config",
extra={
"error": str(e),
"config_file": str(config_file),
},
)
return False

467
src/mcp/cli/cli.py Normal file
View File

@@ -0,0 +1,467 @@
"""MCP CLI tools."""
import importlib.metadata
import importlib.util
import os
import subprocess
import sys
from pathlib import Path
try:
import typer
from typing_extensions import Annotated
except ImportError:
print("Error: typer is required. Install with 'pip install mcp[cli]'")
sys.exit(1)
try:
from mcp.cli import claude
from mcp.server.fastmcp.utilities.logging import get_logger
except ImportError:
print("Error: mcp.server.fastmcp is not installed or not in PYTHONPATH")
sys.exit(1)
try:
import dotenv
except ImportError:
dotenv = None
logger = get_logger("cli")
app = typer.Typer(
name="mcp",
help="MCP development tools",
add_completion=False,
no_args_is_help=True, # Show help if no args provided
)
def _get_npx_command():
"""Get the correct npx command for the current platform."""
if sys.platform == "win32":
# Try both npx.cmd and npx.exe on Windows
for cmd in ["npx.cmd", "npx.exe", "npx"]:
try:
subprocess.run(
[cmd, "--version"], check=True, capture_output=True, shell=True
)
return cmd
except subprocess.CalledProcessError:
continue
return None
return "npx" # On Unix-like systems, just use npx
def _parse_env_var(env_var: str) -> tuple[str, str]:
"""Parse environment variable string in format KEY=VALUE."""
if "=" not in env_var:
logger.error(
f"Invalid environment variable format: {env_var}. Must be KEY=VALUE"
)
sys.exit(1)
key, value = env_var.split("=", 1)
return key.strip(), value.strip()
def _build_uv_command(
file_spec: str,
with_editable: Path | None = None,
with_packages: list[str] | None = None,
) -> list[str]:
"""Build the uv run command that runs a MCP server through mcp run."""
cmd = ["uv"]
cmd.extend(["run", "--with", "mcp"])
if with_editable:
cmd.extend(["--with-editable", str(with_editable)])
if with_packages:
for pkg in with_packages:
if pkg:
cmd.extend(["--with", pkg])
# Add mcp run command
cmd.extend(["mcp", "run", file_spec])
return cmd
def _parse_file_path(file_spec: str) -> tuple[Path, str | None]:
"""Parse a file path that may include a server object specification.
Args:
file_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(file_spec) > 1 and file_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 (file_spec[2:] if has_windows_drive else file_spec):
file_str, server_object = file_spec.rsplit(":", 1)
else:
file_str, server_object = file_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):
"""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
@app.command()
def version() -> None:
"""Show the MCP version."""
try:
version = importlib.metadata.version("mcp")
print(f"MCP version {version}")
except importlib.metadata.PackageNotFoundError:
print("MCP version unknown (package not installed)")
sys.exit(1)
@app.command()
def dev(
file_spec: str = typer.Argument(
...,
help="Python file to run, optionally with :object suffix",
),
with_editable: Annotated[
Path | None,
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with",
help="Additional packages to install",
),
] = [],
) -> None:
"""Run a MCP server with the MCP Inspector."""
file, server_object = _parse_file_path(file_spec)
logger.debug(
"Starting dev server",
extra={
"file": str(file),
"server_object": server_object,
"with_editable": str(with_editable) if with_editable else None,
"with_packages": with_packages,
},
)
try:
# Import server to get dependencies
server = _import_server(file, server_object)
if hasattr(server, "dependencies"):
with_packages = list(set(with_packages + server.dependencies))
uv_cmd = _build_uv_command(file_spec, with_editable, with_packages)
# Get the correct npx command
npx_cmd = _get_npx_command()
if not npx_cmd:
logger.error(
"npx not found. Please ensure Node.js and npm are properly installed "
"and added to your system PATH."
)
sys.exit(1)
# Run the MCP Inspector command with shell=True on Windows
shell = sys.platform == "win32"
process = subprocess.run(
[npx_cmd, "@modelcontextprotocol/inspector"] + uv_cmd,
check=True,
shell=shell,
env=dict(os.environ.items()), # Convert to list of tuples for env update
)
sys.exit(process.returncode)
except subprocess.CalledProcessError as e:
logger.error(
"Dev server failed",
extra={
"file": str(file),
"error": str(e),
"returncode": e.returncode,
},
)
sys.exit(e.returncode)
except FileNotFoundError:
logger.error(
"npx not found. Please ensure Node.js and npm are properly installed "
"and added to your system PATH. You may need to restart your terminal "
"after installation.",
extra={"file": str(file)},
)
sys.exit(1)
@app.command()
def run(
file_spec: str = typer.Argument(
...,
help="Python file to run, optionally with :object suffix",
),
transport: Annotated[
str | None,
typer.Option(
"--transport",
"-t",
help="Transport protocol to use (stdio or sse)",
),
] = None,
) -> None:
"""Run a MCP server.
The server can be specified in two ways:
1. Module approach: server.py - runs the module directly, expecting a server.run() call
2. Import approach: server.py:app - imports and runs the specified server object
Note: This command runs the server directly. You are responsible for ensuring
all dependencies are available. For dependency management, use mcp install
or mcp dev instead.
"""
file, server_object = _parse_file_path(file_spec)
logger.debug(
"Running server",
extra={
"file": str(file),
"server_object": server_object,
"transport": transport,
},
)
try:
# Import and get server object
server = _import_server(file, server_object)
# Run the server
kwargs = {}
if transport:
kwargs["transport"] = transport
server.run(**kwargs)
except Exception as e:
logger.error(
f"Failed to run server: {e}",
extra={
"file": str(file),
"error": str(e),
},
)
sys.exit(1)
@app.command()
def install(
file_spec: str = typer.Argument(
...,
help="Python file to run, optionally with :object suffix",
),
server_name: Annotated[
str | None,
typer.Option(
"--name",
"-n",
help="Custom name for the server (defaults to server's name attribute or file name)",
),
] = None,
with_editable: Annotated[
Path | None,
typer.Option(
"--with-editable",
"-e",
help="Directory containing pyproject.toml to install in editable mode",
exists=True,
file_okay=False,
resolve_path=True,
),
] = None,
with_packages: Annotated[
list[str],
typer.Option(
"--with",
help="Additional packages to install",
),
] = [],
env_vars: Annotated[
list[str],
typer.Option(
"--env-var",
"-e",
help="Environment variables in KEY=VALUE format",
),
] = [],
env_file: Annotated[
Path | None,
typer.Option(
"--env-file",
"-f",
help="Load environment variables from a .env file",
exists=True,
file_okay=True,
dir_okay=False,
resolve_path=True,
),
] = None,
) -> None:
"""Install a MCP server in the Claude desktop app.
Environment variables are preserved once added and only updated if new values
are explicitly provided.
"""
file, server_object = _parse_file_path(file_spec)
logger.debug(
"Installing server",
extra={
"file": str(file),
"server_name": server_name,
"server_object": server_object,
"with_editable": str(with_editable) if with_editable else None,
"with_packages": with_packages,
},
)
if not claude.get_claude_config_path():
logger.error("Claude app not found")
sys.exit(1)
# Try to import server to get its name, but fall back to file name if dependencies missing
name = server_name
server = None
if not name:
try:
server = _import_server(file, server_object)
name = server.name
except (ImportError, ModuleNotFoundError) as e:
logger.debug(
"Could not import server (likely missing dependencies), using file name",
extra={"error": str(e)},
)
name = file.stem
# Get server dependencies if available
server_dependencies = getattr(server, "dependencies", []) if server else []
if server_dependencies:
with_packages = list(set(with_packages + server_dependencies))
# Process environment variables if provided
env_dict: dict[str, str] | None = None
if env_file or env_vars:
env_dict = {}
# Load from .env file if specified
if env_file:
if dotenv:
try:
env_dict |= {
k: v
for k, v in dotenv.dotenv_values(env_file).items()
if v is not None
}
except Exception as e:
logger.error(f"Failed to load .env file: {e}")
sys.exit(1)
else:
logger.error("python-dotenv is not installed. Cannot load .env file.")
sys.exit(1)
# Add command line environment variables
for env_var in env_vars:
key, value = _parse_env_var(env_var)
env_dict[key] = value
if claude.update_claude_config(
file_spec,
name,
with_editable=with_editable,
with_packages=with_packages,
env_vars=env_dict,
):
logger.info(f"Successfully installed {name} in Claude app")
else:
logger.error(f"Failed to install {name} in Claude app")
sys.exit(1)

30
uv.lock generated
View File

@@ -204,6 +204,10 @@ dependencies = [
]
[package.optional-dependencies]
cli = [
{ name = "python-dotenv" },
{ name = "typer" },
]
rich = [
{ name = "rich" },
]
@@ -226,9 +230,11 @@ requires-dist = [
{ name = "httpx-sse", specifier = ">=0.4" },
{ name = "pydantic", specifier = ">=2.7.2,<3.0.0" },
{ name = "pydantic-settings", specifier = ">=2.6.1" },
{ name = "python-dotenv", marker = "extra == 'cli'", specifier = ">=1.0.0" },
{ name = "rich", marker = "extra == 'rich'", specifier = ">=13.9.4" },
{ name = "sse-starlette", specifier = ">=1.6.1" },
{ name = "starlette", specifier = ">=0.27" },
{ name = "typer", marker = "extra == 'cli'", specifier = ">=0.12.4" },
]
[package.metadata.requires-dev]
@@ -605,6 +611,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/3e/14/fd026bc74ded05e2351681545a5f626e78ef831f8edce064d61acd2e6ec7/ruff-0.6.9-py3-none-win_arm64.whl", hash = "sha256:a9641e31476d601f83cd602608739a0840e348bda93fec9f1ee816f8b6798b93", size = 8679879 },
]
[[package]]
name = "shellingham"
version = "1.5.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/58/15/8b3609fd3830ef7b27b655beb4b4e9c62313a4e8da8c676e142cc210d58e/shellingham-1.5.4.tar.gz", hash = "sha256:8dbca0739d487e5bd35ab3ca4b36e11c4078f3a234bfce294b0a0291363404de", size = 10310 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755 },
]
[[package]]
name = "sniffio"
version = "1.3.1"
@@ -704,6 +719,21 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1c/70/efa56ce2271c44a7f4f43533a0477e6854a0948e9f7b76491de1fd3be7c9/trio-0.26.2-py3-none-any.whl", hash = "sha256:c5237e8133eb0a1d72f09a971a55c28ebe69e351c783fc64bc37db8db8bbe1d0", size = 475996 },
]
[[package]]
name = "typer"
version = "0.12.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "click" },
{ name = "rich" },
{ name = "shellingham" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d4/f7/f174a1cae84848ae8b27170a96187b91937b743f0580ff968078fe16930a/typer-0.12.4.tar.gz", hash = "sha256:c9c1613ed6a166162705b3347b8d10b661ccc5d95692654d0fb628118f2c34e6", size = 97945 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ae/cc/15083dcde1252a663398b1b2a173637a3ec65adadfb95137dc95df1e6adc/typer-0.12.4-py3-none-any.whl", hash = "sha256:819aa03699f438397e876aa12b0d63766864ecba1b579092cc9fe35d886e34b6", size = 47402 },
]
[[package]]
name = "typing-extensions"
version = "4.12.2"