mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-27 15:51:33 +02:00
Add Cursor support via CLI integration (#1052)
* Add Cursor support * Use url-safe encoding
This commit is contained in:
Jeremiah Lowin
GitHub
parent
ef742d9400
commit
e50f4ae965
20 changed files with 974 additions and 260 deletions
|
|
@ -10,7 +10,6 @@ import sys
|
|||
from pathlib import Path
|
||||
from typing import Annotated
|
||||
|
||||
import dotenv
|
||||
import typer
|
||||
from pydantic import TypeAdapter
|
||||
from rich.console import Console
|
||||
|
|
@ -18,8 +17,8 @@ from rich.table import Table
|
|||
from typer import Context, Exit
|
||||
|
||||
import fastmcp
|
||||
from fastmcp.cli import claude
|
||||
from fastmcp.cli import run as run_module
|
||||
from fastmcp.cli.install.install import install
|
||||
from fastmcp.server.server import FastMCP
|
||||
from fastmcp.utilities.inspect import FastMCPInfo, inspect_fastmcp
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
|
@ -324,135 +323,8 @@ def run(
|
|||
sys.exit(1)
|
||||
|
||||
|
||||
@app.command()
|
||||
def install(
|
||||
server_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",
|
||||
"-v",
|
||||
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 = run_module.parse_file_path(server_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 = run_module.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:
|
||||
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)
|
||||
|
||||
# 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(
|
||||
server_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)
|
||||
# Add install command directly
|
||||
app.command()(install)
|
||||
|
||||
|
||||
@app.command()
|
||||
|
|
|
|||
5
src/fastmcp/cli/install/__init__.py
Normal file
5
src/fastmcp/cli/install/__init__.py
Normal file
|
|
@ -0,0 +1,5 @@
|
|||
"""Install module for FastMCP CLI."""
|
||||
|
||||
from .install import install
|
||||
|
||||
__all__ = ["install"]
|
||||
121
src/fastmcp/cli/install/claude_desktop.py
Normal file
121
src/fastmcp/cli/install/claude_desktop.py
Normal file
|
|
@ -0,0 +1,121 @@
|
|||
"""Claude Desktop integration for FastMCP install."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from rich import print
|
||||
|
||||
from fastmcp.mcp_config import StdioMCPServer, update_config_file
|
||||
from 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")
|
||||
elif sys.platform.startswith("linux"):
|
||||
path = Path(
|
||||
os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"), "Claude"
|
||||
)
|
||||
else:
|
||||
return None
|
||||
|
||||
if path.exists():
|
||||
return path
|
||||
return None
|
||||
|
||||
|
||||
def install_claude_desktop(
|
||||
server_spec: str,
|
||||
name: str,
|
||||
*,
|
||||
with_editable: Path | None = None,
|
||||
with_packages: list[str] | None = None,
|
||||
env_vars: dict[str, str] | None = None,
|
||||
) -> bool:
|
||||
"""Install FastMCP server in Claude Desktop.
|
||||
|
||||
Args:
|
||||
server_spec: Path to the server file, optionally with :object suffix
|
||||
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
|
||||
|
||||
Returns:
|
||||
True if installation was successful, False otherwise
|
||||
"""
|
||||
config_dir = get_claude_config_path()
|
||||
if not config_dir:
|
||||
print(
|
||||
"[red]❌ Claude Desktop config directory not found.[/red]\n"
|
||||
"[blue]Please ensure Claude Desktop is installed and has been run at least once to initialize its config.[/blue]"
|
||||
)
|
||||
return False
|
||||
|
||||
config_file = config_dir / "claude_desktop_config.json"
|
||||
|
||||
# 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 server_spec:
|
||||
file_path, server_object = server_spec.rsplit(":", 1)
|
||||
server_spec = f"{Path(file_path).resolve()}:{server_object}"
|
||||
else:
|
||||
server_spec = str(Path(server_spec).resolve())
|
||||
|
||||
# Add fastmcp run command
|
||||
args.extend(["fastmcp", "run", server_spec])
|
||||
|
||||
# Create server configuration
|
||||
server_config = StdioMCPServer(
|
||||
command="uv",
|
||||
args=args,
|
||||
env=env_vars or {},
|
||||
)
|
||||
|
||||
try:
|
||||
# Handle environment variable merging manually since we need to preserve existing config
|
||||
if config_file.exists():
|
||||
import json
|
||||
|
||||
content = config_file.read_text().strip()
|
||||
if content:
|
||||
config = json.loads(content)
|
||||
if "mcpServers" in config and name in config["mcpServers"]:
|
||||
existing_env = config["mcpServers"][name].get("env", {})
|
||||
if env_vars:
|
||||
# New vars take precedence over existing ones
|
||||
merged_env = {**existing_env, **env_vars}
|
||||
else:
|
||||
merged_env = existing_env
|
||||
server_config.env = merged_env
|
||||
|
||||
update_config_file(config_file, name, server_config)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(
|
||||
f"[red]Failed to install '[bold]{name}[/bold]' in Claude Desktop: {e}[/red]"
|
||||
)
|
||||
return False
|
||||
135
src/fastmcp/cli/install/cursor.py
Normal file
135
src/fastmcp/cli/install/cursor.py
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
"""Cursor integration for FastMCP install."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from rich import print
|
||||
|
||||
from fastmcp.mcp_config import StdioMCPServer
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
def generate_cursor_deeplink(
|
||||
server_name: str,
|
||||
server_config: StdioMCPServer,
|
||||
) -> str:
|
||||
"""Generate a Cursor deeplink for installing the MCP server.
|
||||
|
||||
Args:
|
||||
server_name: Name of the server
|
||||
server_config: Server configuration
|
||||
|
||||
Returns:
|
||||
Deeplink URL that can be clicked to install the server
|
||||
"""
|
||||
# Create the configuration structure expected by Cursor
|
||||
|
||||
# Base64 encode the configuration (URL-safe for query parameter)
|
||||
config_json = server_config.model_dump_json(exclude_none=True)
|
||||
config_b64 = base64.urlsafe_b64encode(config_json.encode()).decode()
|
||||
|
||||
# Generate the deeplink URL
|
||||
deeplink = f"cursor://anysphere.cursor-deeplink/mcp/install?name={server_name}&config={config_b64}"
|
||||
|
||||
return deeplink
|
||||
|
||||
|
||||
def open_deeplink(deeplink: str) -> bool:
|
||||
"""Attempt to open a deeplink URL using the system's default handler.
|
||||
|
||||
Args:
|
||||
deeplink: The deeplink URL to open
|
||||
|
||||
Returns:
|
||||
True if the command succeeded, False otherwise
|
||||
"""
|
||||
try:
|
||||
if sys.platform == "darwin": # macOS
|
||||
subprocess.run(["open", deeplink], check=True, capture_output=True)
|
||||
elif sys.platform == "win32": # Windows
|
||||
subprocess.run(
|
||||
["start", deeplink], shell=True, check=True, capture_output=True
|
||||
)
|
||||
else: # Linux and others
|
||||
subprocess.run(["xdg-open", deeplink], check=True, capture_output=True)
|
||||
return True
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
return False
|
||||
|
||||
|
||||
def install_cursor(
|
||||
server_spec: str,
|
||||
name: str,
|
||||
*,
|
||||
with_editable: Path | None = None,
|
||||
with_packages: list[str] | None = None,
|
||||
env_vars: dict[str, str] | None = None,
|
||||
) -> bool:
|
||||
"""Install FastMCP server in Cursor.
|
||||
|
||||
Args:
|
||||
server_spec: Path to the server file, optionally with :object suffix
|
||||
name: Name for the server in Cursor'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
|
||||
|
||||
Returns:
|
||||
True if installation was successful, False otherwise
|
||||
"""
|
||||
# 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 server_spec:
|
||||
file_path, server_object = server_spec.rsplit(":", 1)
|
||||
server_spec = f"{Path(file_path).resolve()}:{server_object}"
|
||||
else:
|
||||
server_spec = str(Path(server_spec).resolve())
|
||||
|
||||
# Add fastmcp run command
|
||||
args.extend(["fastmcp", "run", server_spec])
|
||||
|
||||
# Create server configuration
|
||||
server_config = StdioMCPServer(
|
||||
command="uv",
|
||||
args=args,
|
||||
env=env_vars or {},
|
||||
)
|
||||
|
||||
# Generate and open deeplink
|
||||
try:
|
||||
deeplink = generate_cursor_deeplink(name, server_config)
|
||||
|
||||
if open_deeplink(deeplink):
|
||||
print(
|
||||
f"[green]Opening Cursor to install '[bold]{name}[/bold]' - please confirm in Cursor to complete installation[/green]"
|
||||
)
|
||||
return True
|
||||
else:
|
||||
print("[yellow]Could not open Cursor automatically.[/yellow]")
|
||||
print(f"[blue]Please open this link to install:[/blue] {deeplink}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"[red]Failed to generate Cursor deeplink: {e}[/red]")
|
||||
return False
|
||||
185
src/fastmcp/cli/install/install.py
Normal file
185
src/fastmcp/cli/install/install.py
Normal file
|
|
@ -0,0 +1,185 @@
|
|||
"""Main install logic for FastMCP CLI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Annotated
|
||||
|
||||
import typer
|
||||
from dotenv import dotenv_values
|
||||
from rich import print
|
||||
|
||||
from fastmcp.cli.run import import_server, parse_file_path
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
|
||||
from .claude_desktop import install_claude_desktop
|
||||
from .cursor import install_cursor
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class Client(str, Enum):
|
||||
"""Supported MCP clients."""
|
||||
|
||||
CLAUDE_DESKTOP = "claude-desktop"
|
||||
CURSOR = "cursor"
|
||||
|
||||
|
||||
def install(
|
||||
client: Annotated[
|
||||
Client,
|
||||
typer.Argument(help="MCP client to install the server into"),
|
||||
],
|
||||
server_spec: Annotated[
|
||||
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. Use this to include local packages that are not available on PyPI.",
|
||||
exists=True,
|
||||
file_okay=False,
|
||||
resolve_path=True,
|
||||
),
|
||||
] = None,
|
||||
with_packages: Annotated[
|
||||
list[str],
|
||||
typer.Option(
|
||||
"--with",
|
||||
help="Additional packages to install, in PEP 508 format (e.g. 'httpx>=0.25.2')",
|
||||
),
|
||||
] = [],
|
||||
env_vars: Annotated[
|
||||
list[str],
|
||||
typer.Option(
|
||||
"--env-var",
|
||||
"-v",
|
||||
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 specified target application.
|
||||
|
||||
Environment variables are preserved once added and only updated if new values
|
||||
are explicitly provided.
|
||||
"""
|
||||
|
||||
# Parse server spec
|
||||
file, server_object = parse_file_path(server_spec)
|
||||
|
||||
logger.debug(
|
||||
"Installing server",
|
||||
extra={
|
||||
"client": client,
|
||||
"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,
|
||||
},
|
||||
)
|
||||
|
||||
# Try to import server to get its name and dependencies
|
||||
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:
|
||||
try:
|
||||
env_dict |= {
|
||||
k: v for k, v in dotenv_values(env_file).items() if v is not None
|
||||
}
|
||||
except Exception as e:
|
||||
print(f"[red]❌ Failed to load .env file: {e}[/red]")
|
||||
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
|
||||
|
||||
# Route to appropriate installer
|
||||
if client == Client.CLAUDE_DESKTOP:
|
||||
success = install_claude_desktop(
|
||||
server_spec=server_spec,
|
||||
name=name,
|
||||
with_editable=with_editable,
|
||||
with_packages=with_packages,
|
||||
env_vars=env_dict,
|
||||
)
|
||||
elif client == Client.CURSOR:
|
||||
success = install_cursor(
|
||||
server_spec=server_spec,
|
||||
name=name,
|
||||
with_editable=with_editable,
|
||||
with_packages=with_packages,
|
||||
env_vars=env_dict,
|
||||
)
|
||||
else:
|
||||
print(
|
||||
f"[red bold]Unknown client: {client!r}[/red bold]. Supported clients: [bold]{Client.CLAUDE_DESKTOP}[/bold], [bold]{Client.CURSOR}[/bold]"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
if success:
|
||||
# Only show generic success message for clients that don't have their own messaging
|
||||
if client != Client.CURSOR:
|
||||
print(
|
||||
f"[green bold]Successfully installed '[bold]{name}[/bold]' in {client.value}[/green bold]"
|
||||
)
|
||||
else:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _parse_env_var(env_var: str) -> tuple[str, str]:
|
||||
"""Parse environment variable string in format KEY=VALUE."""
|
||||
if "=" not in env_var:
|
||||
print(
|
||||
f"[red]❌ Invalid environment variable format: '[bold]{env_var}[/bold]'. Must be KEY=VALUE[/red]"
|
||||
)
|
||||
sys.exit(1)
|
||||
key, value = env_var.split("=", 1)
|
||||
return key.strip(), value.strip()
|
||||
|
|
@ -31,11 +31,11 @@ from fastmcp.client.roots import (
|
|||
)
|
||||
from fastmcp.client.sampling import SamplingHandler, create_sampling_callback
|
||||
from fastmcp.exceptions import ToolError
|
||||
from fastmcp.mcp_config import MCPConfig
|
||||
from fastmcp.server import FastMCP
|
||||
from fastmcp.utilities.exceptions import get_catch_handlers
|
||||
from fastmcp.utilities.json_schema_type import json_schema_to_type
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.mcp_config import MCPConfig
|
||||
from fastmcp.utilities.types import get_cached_typeadapter
|
||||
|
||||
from .transports import (
|
||||
|
|
|
|||
|
|
@ -29,10 +29,10 @@ from typing_extensions import TypedDict, Unpack
|
|||
import fastmcp
|
||||
from fastmcp.client.auth.bearer import BearerAuth
|
||||
from fastmcp.client.auth.oauth import OAuth
|
||||
from fastmcp.mcp_config import MCPConfig, infer_transport_type_from_url
|
||||
from fastmcp.server.dependencies import get_http_headers
|
||||
from fastmcp.server.server import FastMCP
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.mcp_config import MCPConfig, infer_transport_type_from_url
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
|
|
|||
282
src/fastmcp/mcp_config.py
Normal file
282
src/fastmcp/mcp_config.py
Normal file
|
|
@ -0,0 +1,282 @@
|
|||
"""Canonical MCP Configuration Format.
|
||||
|
||||
This module defines the standard configuration format for Model Context Protocol (MCP) servers.
|
||||
It provides a client-agnostic, extensible format that can be used across all MCP implementations.
|
||||
|
||||
The configuration format supports both stdio and remote (HTTP/SSE) transports, with comprehensive
|
||||
field definitions for server metadata, authentication, and execution parameters.
|
||||
|
||||
Example configuration:
|
||||
{
|
||||
"mcpServers": {
|
||||
"my-server": {
|
||||
"command": "npx",
|
||||
"args": ["-y", "@my/mcp-server"],
|
||||
"env": {"API_KEY": "secret"},
|
||||
"timeout": 30000,
|
||||
"description": "My MCP server"
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Annotated, Any, Literal
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx
|
||||
from pydantic import AnyUrl, BaseModel, ConfigDict, Field
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.client.transports import (
|
||||
SSETransport,
|
||||
StdioTransport,
|
||||
StreamableHttpTransport,
|
||||
)
|
||||
|
||||
|
||||
def infer_transport_type_from_url(
|
||||
url: str | AnyUrl,
|
||||
) -> Literal["http", "sse"]:
|
||||
"""
|
||||
Infer the appropriate transport type from the given URL.
|
||||
"""
|
||||
url = str(url)
|
||||
if not url.startswith("http"):
|
||||
raise ValueError(f"Invalid URL: {url}")
|
||||
|
||||
parsed_url = urlparse(url)
|
||||
path = parsed_url.path
|
||||
|
||||
# Match /sse followed by /, ?, &, or end of string
|
||||
if re.search(r"/sse(/|\?|&|$)", path):
|
||||
return "sse"
|
||||
else:
|
||||
return "http"
|
||||
|
||||
|
||||
class StdioMCPServer(BaseModel):
|
||||
"""MCP server configuration for stdio transport.
|
||||
|
||||
This is the canonical configuration format for MCP servers using stdio transport.
|
||||
"""
|
||||
|
||||
# Required fields
|
||||
command: str
|
||||
|
||||
# Common optional fields
|
||||
args: list[str] = Field(default_factory=list)
|
||||
env: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
# Transport specification
|
||||
transport: Literal["stdio"] = "stdio"
|
||||
type: Literal["stdio"] | None = None # Alternative transport field name
|
||||
|
||||
# Execution context
|
||||
cwd: str | None = None # Working directory for command execution
|
||||
timeout: int | None = None # Maximum response time in milliseconds
|
||||
|
||||
# Metadata
|
||||
description: str | None = None # Human-readable server description
|
||||
icon: str | None = None # Icon path or URL for UI display
|
||||
|
||||
# Authentication configuration
|
||||
authentication: dict[str, Any] | None = None # Auth configuration object
|
||||
|
||||
model_config = ConfigDict(extra="allow") # Preserve unknown fields
|
||||
|
||||
def to_transport(self) -> StdioTransport:
|
||||
from fastmcp.client.transports import StdioTransport
|
||||
|
||||
return StdioTransport(
|
||||
command=self.command,
|
||||
args=self.args,
|
||||
env=self.env,
|
||||
cwd=self.cwd,
|
||||
)
|
||||
|
||||
|
||||
class RemoteMCPServer(BaseModel):
|
||||
"""MCP server configuration for HTTP/SSE transport.
|
||||
|
||||
This is the canonical configuration format for MCP servers using remote transports.
|
||||
"""
|
||||
|
||||
# Required fields
|
||||
url: str
|
||||
|
||||
# Transport configuration
|
||||
transport: Literal["http", "streamable-http", "sse"] | None = None
|
||||
headers: dict[str, str] = Field(default_factory=dict)
|
||||
|
||||
# Authentication
|
||||
auth: Annotated[
|
||||
str | Literal["oauth"] | httpx.Auth | None,
|
||||
Field(
|
||||
description='Either a string representing a Bearer token, the literal "oauth" to use OAuth authentication, or an httpx.Auth instance for custom authentication.',
|
||||
),
|
||||
] = None
|
||||
|
||||
# Timeout configuration
|
||||
sse_read_timeout: datetime.timedelta | int | float | None = None
|
||||
timeout: int | None = None # Maximum response time in milliseconds
|
||||
|
||||
# Metadata
|
||||
description: str | None = None # Human-readable server description
|
||||
icon: str | None = None # Icon path or URL for UI display
|
||||
|
||||
# Authentication configuration
|
||||
authentication: dict[str, Any] | None = None # Auth configuration object
|
||||
|
||||
model_config = ConfigDict(
|
||||
extra="allow", arbitrary_types_allowed=True
|
||||
) # Preserve unknown fields
|
||||
|
||||
def to_transport(self) -> StreamableHttpTransport | SSETransport:
|
||||
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
|
||||
|
||||
if self.transport is None:
|
||||
transport = infer_transport_type_from_url(self.url)
|
||||
else:
|
||||
transport = self.transport
|
||||
|
||||
if transport == "sse":
|
||||
return SSETransport(
|
||||
self.url,
|
||||
headers=self.headers,
|
||||
auth=self.auth,
|
||||
sse_read_timeout=self.sse_read_timeout,
|
||||
)
|
||||
else:
|
||||
# Both "http" and "streamable-http" map to StreamableHttpTransport
|
||||
return StreamableHttpTransport(
|
||||
self.url,
|
||||
headers=self.headers,
|
||||
auth=self.auth,
|
||||
sse_read_timeout=self.sse_read_timeout,
|
||||
)
|
||||
|
||||
|
||||
class MCPConfig(BaseModel):
|
||||
"""Canonical MCP configuration format.
|
||||
|
||||
This defines the standard configuration format for Model Context Protocol servers.
|
||||
The format is designed to be client-agnostic and extensible for future use cases.
|
||||
"""
|
||||
|
||||
mcpServers: dict[str, StdioMCPServer | RemoteMCPServer]
|
||||
|
||||
model_config = ConfigDict(extra="allow") # Preserve unknown top-level fields
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, config: dict[str, Any]) -> MCPConfig:
|
||||
"""Parse MCP configuration from dictionary format."""
|
||||
# Handle case where config is just the mcpServers object
|
||||
if "mcpServers" not in config and any(
|
||||
isinstance(v, dict) and ("command" in v or "url" in v)
|
||||
for v in config.values()
|
||||
):
|
||||
# This looks like a bare mcpServers object
|
||||
servers_dict = config
|
||||
else:
|
||||
# Standard format with mcpServers wrapper
|
||||
servers_dict = config.get("mcpServers", {})
|
||||
|
||||
# Parse each server configuration
|
||||
parsed_servers = {}
|
||||
for name, server_config in servers_dict.items():
|
||||
if not isinstance(server_config, dict):
|
||||
continue
|
||||
|
||||
# Determine if this is stdio or remote based on fields
|
||||
if "command" in server_config:
|
||||
parsed_servers[name] = StdioMCPServer.model_validate(server_config)
|
||||
elif "url" in server_config:
|
||||
parsed_servers[name] = RemoteMCPServer.model_validate(server_config)
|
||||
else:
|
||||
# Skip invalid server configs but preserve them as raw dicts
|
||||
# This allows for forward compatibility with unknown server types
|
||||
continue
|
||||
|
||||
# Create config with any extra top-level fields preserved
|
||||
config_data = {k: v for k, v in config.items() if k != "mcpServers"}
|
||||
config_data["mcpServers"] = parsed_servers
|
||||
|
||||
return cls.model_validate(config_data)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Convert MCPConfig to dictionary format, preserving all fields."""
|
||||
# Start with all extra fields at the top level
|
||||
result = self.model_dump(exclude={"mcpServers"}, exclude_none=True)
|
||||
|
||||
# Add mcpServers with all fields preserved
|
||||
result["mcpServers"] = {
|
||||
name: server.model_dump(exclude_none=True)
|
||||
for name, server in self.mcpServers.items()
|
||||
}
|
||||
|
||||
return result
|
||||
|
||||
def write_to_file(self, file_path: Path) -> None:
|
||||
"""Write configuration to JSON file."""
|
||||
file_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(file_path, "w") as f:
|
||||
json.dump(self.to_dict(), f, indent=2)
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, file_path: Path) -> MCPConfig:
|
||||
"""Load configuration from JSON file."""
|
||||
if not file_path.exists():
|
||||
return cls(mcpServers={})
|
||||
with open(file_path) as f:
|
||||
content = f.read().strip()
|
||||
if not content:
|
||||
return cls(mcpServers={})
|
||||
data = json.loads(content)
|
||||
return cls.from_dict(data)
|
||||
|
||||
def add_server(self, name: str, server: StdioMCPServer | RemoteMCPServer) -> None:
|
||||
"""Add or update a server in the configuration."""
|
||||
self.mcpServers[name] = server
|
||||
|
||||
def remove_server(self, name: str) -> None:
|
||||
"""Remove a server from the configuration."""
|
||||
if name in self.mcpServers:
|
||||
del self.mcpServers[name]
|
||||
|
||||
|
||||
def update_config_file(
|
||||
file_path: Path,
|
||||
server_name: str,
|
||||
server_config: StdioMCPServer | RemoteMCPServer,
|
||||
) -> None:
|
||||
"""Update MCP configuration file with new server, preserving existing fields."""
|
||||
config = MCPConfig.from_file(file_path)
|
||||
|
||||
# If updating an existing server, merge with existing configuration
|
||||
# to preserve any unknown fields
|
||||
if server_name in config.mcpServers:
|
||||
existing_server = config.mcpServers[server_name]
|
||||
# Get the raw dict representation of both servers
|
||||
existing_dict = existing_server.model_dump()
|
||||
new_dict = server_config.model_dump(exclude_none=True)
|
||||
|
||||
# Merge, with new values taking precedence
|
||||
merged_dict = {**existing_dict, **new_dict}
|
||||
|
||||
# Create new server instance with merged data
|
||||
if "command" in merged_dict:
|
||||
merged_server = StdioMCPServer.model_validate(merged_dict)
|
||||
else:
|
||||
merged_server = RemoteMCPServer.model_validate(merged_dict)
|
||||
|
||||
config.add_server(server_name, merged_server)
|
||||
else:
|
||||
config.add_server(server_name, server_config)
|
||||
|
||||
config.write_to_file(file_path)
|
||||
|
|
@ -22,6 +22,7 @@ from fastmcp.client.logging import LogMessage
|
|||
from fastmcp.client.roots import RootsList
|
||||
from fastmcp.client.transports import ClientTransportT
|
||||
from fastmcp.exceptions import NotFoundError, ResourceError, ToolError
|
||||
from fastmcp.mcp_config import MCPConfig
|
||||
from fastmcp.prompts import Prompt, PromptMessage
|
||||
from fastmcp.prompts.prompt import PromptArgument
|
||||
from fastmcp.prompts.prompt_manager import PromptManager
|
||||
|
|
@ -33,7 +34,6 @@ from fastmcp.server.server import FastMCP
|
|||
from fastmcp.tools.tool import Tool, ToolResult
|
||||
from fastmcp.tools.tool_manager import ToolManager
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.mcp_config import MCPConfig
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.server import Context
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ from starlette.routing import BaseRoute, Route
|
|||
import fastmcp
|
||||
import fastmcp.server
|
||||
from fastmcp.exceptions import DisabledError, NotFoundError
|
||||
from fastmcp.mcp_config import MCPConfig
|
||||
from fastmcp.prompts import Prompt, PromptManager
|
||||
from fastmcp.prompts.prompt import FunctionPrompt
|
||||
from fastmcp.resources import Resource, ResourceManager
|
||||
|
|
@ -63,7 +64,6 @@ from fastmcp.utilities.cache import TimedCache
|
|||
from fastmcp.utilities.cli import log_server_banner
|
||||
from fastmcp.utilities.components import FastMCPComponent
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.mcp_config import MCPConfig
|
||||
from fastmcp.utilities.types import NotSet, NotSetT
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
|
|||
|
|
@ -1,103 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Annotated, Any, Literal
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx
|
||||
from pydantic import AnyUrl, ConfigDict, Field
|
||||
|
||||
from fastmcp.utilities.types import FastMCPBaseModel
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from fastmcp.client.transports import (
|
||||
SSETransport,
|
||||
StdioTransport,
|
||||
StreamableHttpTransport,
|
||||
)
|
||||
|
||||
|
||||
def infer_transport_type_from_url(
|
||||
url: str | AnyUrl,
|
||||
) -> Literal["http", "sse"]:
|
||||
"""
|
||||
Infer the appropriate transport type from the given URL.
|
||||
"""
|
||||
url = str(url)
|
||||
if not url.startswith("http"):
|
||||
raise ValueError(f"Invalid URL: {url}")
|
||||
|
||||
parsed_url = urlparse(url)
|
||||
path = parsed_url.path
|
||||
|
||||
# Match /sse followed by /, ?, &, or end of string
|
||||
if re.search(r"/sse(/|\?|&|$)", path):
|
||||
return "sse"
|
||||
else:
|
||||
return "http"
|
||||
|
||||
|
||||
class StdioMCPServer(FastMCPBaseModel):
|
||||
command: str
|
||||
args: list[str] = Field(default_factory=list)
|
||||
env: dict[str, Any] = Field(default_factory=dict)
|
||||
cwd: str | None = None
|
||||
transport: Literal["stdio"] = "stdio"
|
||||
|
||||
def to_transport(self) -> StdioTransport:
|
||||
from fastmcp.client.transports import StdioTransport
|
||||
|
||||
return StdioTransport(
|
||||
command=self.command,
|
||||
args=self.args,
|
||||
env=self.env,
|
||||
cwd=self.cwd,
|
||||
)
|
||||
|
||||
|
||||
class RemoteMCPServer(FastMCPBaseModel):
|
||||
url: str
|
||||
headers: dict[str, str] = Field(default_factory=dict)
|
||||
transport: Literal["http", "streamable-http", "sse"] | None = None
|
||||
auth: Annotated[
|
||||
str | Literal["oauth"] | httpx.Auth | None,
|
||||
Field(
|
||||
description='Either a string representing a Bearer token, the literal "oauth" to use OAuth authentication, or an httpx.Auth instance for custom authentication.',
|
||||
),
|
||||
] = None
|
||||
sse_read_timeout: datetime.timedelta | int | float | None = None
|
||||
|
||||
model_config = ConfigDict(arbitrary_types_allowed=True)
|
||||
|
||||
def to_transport(self) -> StreamableHttpTransport | SSETransport:
|
||||
from fastmcp.client.transports import SSETransport, StreamableHttpTransport
|
||||
|
||||
if self.transport is None:
|
||||
transport = infer_transport_type_from_url(self.url)
|
||||
else:
|
||||
transport = self.transport
|
||||
|
||||
if transport == "sse":
|
||||
return SSETransport(
|
||||
self.url,
|
||||
headers=self.headers,
|
||||
auth=self.auth,
|
||||
sse_read_timeout=self.sse_read_timeout,
|
||||
)
|
||||
else:
|
||||
# Both "http" and "streamable-http" map to StreamableHttpTransport
|
||||
return StreamableHttpTransport(
|
||||
self.url,
|
||||
headers=self.headers,
|
||||
auth=self.auth,
|
||||
sse_read_timeout=self.sse_read_timeout,
|
||||
)
|
||||
|
||||
|
||||
class MCPConfig(FastMCPBaseModel):
|
||||
mcpServers: dict[str, StdioMCPServer | RemoteMCPServer]
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, config: dict[str, Any]) -> MCPConfig:
|
||||
return cls(mcpServers=config.get("mcpServers", config))
|
||||
Loading…
Add table
Add a link
Reference in a new issue