mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Compare commits
1 commit
main
...
fastmcp-ge
| Author | SHA1 | Date | |
|---|---|---|---|
|
fb28f9cda9 |
3 changed files with 1108 additions and 0 deletions
|
|
@ -10,6 +10,7 @@ import sys
|
|||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Literal
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import cyclopts
|
||||
import pyperclip
|
||||
|
|
@ -17,10 +18,17 @@ from rich.console import Console
|
|||
from rich.table import Table
|
||||
|
||||
import fastmcp
|
||||
from fastmcp import Client
|
||||
from fastmcp.cli import run as run_module
|
||||
from fastmcp.cli.install import install_app
|
||||
from fastmcp.client.auth import OAuth
|
||||
from fastmcp.server.server import FastMCP
|
||||
from fastmcp.utilities.cli import is_already_in_uv_subprocess, load_and_merge_config
|
||||
from fastmcp.utilities.generate import (
|
||||
generate_agents_md,
|
||||
generate_tool_script,
|
||||
to_snake_case,
|
||||
)
|
||||
from fastmcp.utilities.inspect import (
|
||||
InspectFormat,
|
||||
format_info,
|
||||
|
|
@ -867,6 +875,186 @@ async def prepare(
|
|||
sys.exit(1)
|
||||
|
||||
|
||||
def _infer_server_name(url: str) -> str:
|
||||
"""Infer a server name from a URL.
|
||||
|
||||
Args:
|
||||
url: Server URL
|
||||
|
||||
Returns:
|
||||
Server name suitable for use as a directory name
|
||||
"""
|
||||
parsed = urlparse(url)
|
||||
# Use hostname or last path component
|
||||
if parsed.hostname:
|
||||
# Remove common prefixes and suffixes
|
||||
name = (
|
||||
parsed.hostname.replace("www.", "").replace(".com", "").replace(".org", "")
|
||||
)
|
||||
# If there's a path, use the last component
|
||||
if parsed.path and parsed.path != "/":
|
||||
path_parts = [p for p in parsed.path.split("/") if p]
|
||||
if path_parts:
|
||||
name = path_parts[-1]
|
||||
else:
|
||||
# Fallback to using part of the URL
|
||||
name = "mcp_server"
|
||||
|
||||
return to_snake_case(name)
|
||||
|
||||
|
||||
@app.command
|
||||
async def generate(
|
||||
url: str,
|
||||
*,
|
||||
output: Annotated[
|
||||
Path | None,
|
||||
cyclopts.Parameter(
|
||||
name=["--output", "-o"],
|
||||
help="Output directory for generated scripts",
|
||||
),
|
||||
] = None,
|
||||
auth: Annotated[
|
||||
str | None,
|
||||
cyclopts.Parameter(
|
||||
"--auth",
|
||||
help='Authentication: "oauth" for OAuth, "$VAR" for env var, or literal token',
|
||||
),
|
||||
] = None,
|
||||
server_name: Annotated[
|
||||
str | None,
|
||||
cyclopts.Parameter(
|
||||
"--server-name",
|
||||
help="Override server name",
|
||||
),
|
||||
] = None,
|
||||
) -> None:
|
||||
"""Generate standalone Python scripts from MCP server tools.
|
||||
|
||||
Connects to an MCP server and generates a directory of Python scripts,
|
||||
one per tool, that agents can discover and use progressively without
|
||||
loading all tool definitions into context.
|
||||
|
||||
Examples:
|
||||
# Basic usage (no auth)
|
||||
fastmcp generate https://mcp.example.com/mcp
|
||||
|
||||
# With OAuth
|
||||
fastmcp generate https://mcp.example.com/mcp --auth oauth
|
||||
|
||||
# With environment variable
|
||||
export MY_API_TOKEN="secret"
|
||||
fastmcp generate https://mcp.example.com/mcp --auth '$MY_API_TOKEN'
|
||||
|
||||
# With embedded token (hardcoded in scripts)
|
||||
fastmcp generate https://mcp.example.com/mcp --auth "sk-secret-token"
|
||||
|
||||
# Custom output directory
|
||||
fastmcp generate https://mcp.example.com/mcp --output ./my_tools
|
||||
|
||||
Args:
|
||||
url: URL of the MCP server to connect to
|
||||
"""
|
||||
logger.debug(
|
||||
"Generating code from MCP server",
|
||||
extra={
|
||||
"url": url,
|
||||
"output": str(output) if output else None,
|
||||
"auth": bool(auth),
|
||||
},
|
||||
)
|
||||
|
||||
# Parse authentication and determine mode
|
||||
auth_obj = None
|
||||
auth_mode = "none"
|
||||
auth_value = None
|
||||
|
||||
if auth:
|
||||
if auth == "oauth":
|
||||
# OAuth mode
|
||||
auth_obj = OAuth(mcp_url=url)
|
||||
auth_mode = "oauth"
|
||||
logger.debug("Using OAuth authentication")
|
||||
elif auth.startswith("$"):
|
||||
# Environment variable mode
|
||||
env_var_name = auth[1:] # Strip the $
|
||||
token = os.environ.get(env_var_name)
|
||||
if not token:
|
||||
console.print(
|
||||
f"[red]✗[/red] Environment variable {env_var_name} is not set"
|
||||
)
|
||||
sys.exit(1)
|
||||
auth_obj = token
|
||||
auth_mode = "env_var"
|
||||
auth_value = env_var_name
|
||||
logger.debug(f"Using token from environment variable: {env_var_name}")
|
||||
else:
|
||||
# Literal token mode
|
||||
auth_obj = auth
|
||||
auth_mode = "token"
|
||||
auth_value = auth
|
||||
logger.debug("Using embedded token authentication")
|
||||
|
||||
# Connect to server, list tools, and generate scripts
|
||||
try:
|
||||
console.print(f"[cyan]Connecting to[/cyan] {url} ...")
|
||||
async with Client(url, auth=auth_obj) as client:
|
||||
tools = await client.list_tools()
|
||||
|
||||
# Get server name and instructions from the server's initialization result
|
||||
server_instructions = None
|
||||
if server_name is None:
|
||||
if client.initialize_result and client.initialize_result.serverInfo:
|
||||
server_name = to_snake_case(
|
||||
client.initialize_result.serverInfo.name
|
||||
)
|
||||
else:
|
||||
server_name = _infer_server_name(url)
|
||||
|
||||
# Extract server instructions if available
|
||||
if client.initialize_result and client.initialize_result.instructions:
|
||||
server_instructions = client.initialize_result.instructions
|
||||
|
||||
if output is None:
|
||||
output = Path(server_name)
|
||||
|
||||
if not tools:
|
||||
console.print("[yellow]⚠[/yellow] No tools found on server")
|
||||
return
|
||||
|
||||
console.print(f"[green]✓[/green] Found {len(tools)} tools")
|
||||
|
||||
# Generate tool scripts
|
||||
output.mkdir(parents=True, exist_ok=True)
|
||||
console.print(f"[cyan]Generating scripts in[/cyan] {output}/")
|
||||
|
||||
for tool in tools:
|
||||
script = generate_tool_script(tool, url, auth_mode, auth_value)
|
||||
filename = to_snake_case(tool.name) + ".py"
|
||||
script_path = output / filename
|
||||
script_path.write_text(script)
|
||||
logger.debug(f"Generated script: {filename}")
|
||||
|
||||
# Generate metadata files
|
||||
agents_md = generate_agents_md(
|
||||
server_name, url, tools, auth_mode, auth_value, server_instructions
|
||||
)
|
||||
(output / "AGENTS.md").write_text(agents_md)
|
||||
|
||||
console.print(f"[green]✓[/green] Generated {len(tools)} tool scripts")
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(
|
||||
"Failed to connect to MCP server or generate scripts",
|
||||
extra={
|
||||
"url": url,
|
||||
"error": str(e),
|
||||
},
|
||||
)
|
||||
console.print(f"[red]✗[/red] Failed: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# Add project subcommand group
|
||||
app.command(project_app)
|
||||
|
||||
|
|
|
|||
310
src/fastmcp/utilities/generate.py
Normal file
310
src/fastmcp/utilities/generate.py
Normal file
|
|
@ -0,0 +1,310 @@
|
|||
"""Code generation utilities for MCP tools.
|
||||
|
||||
This module provides functions to generate standalone Python scripts from MCP tool
|
||||
definitions, enabling progressive discovery and context-efficient agent workflows.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import mcp.types
|
||||
|
||||
|
||||
def to_snake_case(name: str) -> str:
|
||||
"""Convert a tool name to snake_case for use as a Python identifier.
|
||||
|
||||
Args:
|
||||
name: Tool name (e.g., "get-document", "getTabs", "list.items")
|
||||
|
||||
Returns:
|
||||
Snake case identifier (e.g., "get_document", "get_tabs", "list_items")
|
||||
"""
|
||||
# Replace common separators with underscores
|
||||
result = name.replace("-", "_").replace(".", "_").replace(" ", "_")
|
||||
# Handle camelCase by inserting underscores before capitals
|
||||
import re
|
||||
|
||||
result = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", result)
|
||||
return result.lower()
|
||||
|
||||
|
||||
def json_schema_to_python_type(prop: dict) -> str:
|
||||
"""Convert JSON schema property to Python type hint.
|
||||
|
||||
Args:
|
||||
prop: JSON schema property definition
|
||||
|
||||
Returns:
|
||||
Python type hint string (e.g., "str", "int", "list", "dict")
|
||||
"""
|
||||
json_type = prop.get("type", "any")
|
||||
|
||||
type_map = {
|
||||
"string": "str",
|
||||
"integer": "int",
|
||||
"number": "float",
|
||||
"boolean": "bool",
|
||||
"array": "list",
|
||||
"object": "dict",
|
||||
}
|
||||
|
||||
return type_map.get(json_type, "Any")
|
||||
|
||||
|
||||
def generate_typed_params(input_schema: dict) -> tuple[str, list[str]]:
|
||||
"""Generate function parameters and parameter names from JSON schema.
|
||||
|
||||
Args:
|
||||
input_schema: JSON schema for tool input parameters
|
||||
|
||||
Returns:
|
||||
Tuple of (params_str, param_names) where:
|
||||
- params_str: Formatted function parameters (e.g., "name: str, age: int | None = None")
|
||||
- param_names: List of parameter names for building the args dict
|
||||
"""
|
||||
properties = input_schema.get("properties", {})
|
||||
required = input_schema.get("required", [])
|
||||
|
||||
params = []
|
||||
param_names = []
|
||||
|
||||
for name, prop in properties.items():
|
||||
param_names.append(name)
|
||||
python_type = json_schema_to_python_type(prop)
|
||||
|
||||
if name in required:
|
||||
params.append(f"{name}: {python_type}")
|
||||
else:
|
||||
params.append(f"{name}: {python_type} | None = None")
|
||||
|
||||
return ", ".join(params), param_names
|
||||
|
||||
|
||||
def generate_args_dict(param_names: list[str], indent: str = " ") -> str:
|
||||
"""Generate the arguments dictionary for tool calling.
|
||||
|
||||
Args:
|
||||
param_names: List of parameter names
|
||||
indent: Indentation string for formatting
|
||||
|
||||
Returns:
|
||||
Formatted dictionary string for passing to call_tool
|
||||
"""
|
||||
if not param_names:
|
||||
return "{}"
|
||||
|
||||
lines = ["{"]
|
||||
for name in param_names:
|
||||
lines.append(f'{indent}"{name}": {name},')
|
||||
lines.append(f"{indent[:-4]}}}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def generate_auth_code(
|
||||
auth_mode: str, auth_value: str | None, server_url: str
|
||||
) -> tuple[str, str]:
|
||||
"""Generate authentication code for the tool script.
|
||||
|
||||
Args:
|
||||
auth_mode: Authentication mode ("none", "oauth", "env_var", "token")
|
||||
auth_value: Auth value (env var name for env_var mode, token for token mode)
|
||||
server_url: URL of the MCP server (for OAuth)
|
||||
|
||||
Returns:
|
||||
Tuple of (imports, get_auth_function) where:
|
||||
- imports: Import statements needed for auth
|
||||
- get_auth_function: Complete get_auth() function implementation
|
||||
"""
|
||||
if auth_mode == "oauth":
|
||||
imports = "from fastmcp.client.auth import OAuth"
|
||||
get_auth = f'''def get_auth():
|
||||
"""Get authentication for the MCP server."""
|
||||
return OAuth(mcp_url="{server_url}")'''
|
||||
|
||||
elif auth_mode == "env_var":
|
||||
imports = ""
|
||||
get_auth = f'''def get_auth():
|
||||
"""Get authentication for the MCP server."""
|
||||
token = os.environ.get("{auth_value}")
|
||||
if not token:
|
||||
raise ValueError("Missing required environment variable: {auth_value}")
|
||||
return token'''
|
||||
|
||||
elif auth_mode == "token":
|
||||
imports = ""
|
||||
get_auth = f'''def get_auth():
|
||||
"""Get authentication for the MCP server."""
|
||||
return "{auth_value}"'''
|
||||
|
||||
else: # none
|
||||
imports = ""
|
||||
get_auth = '''def get_auth():
|
||||
"""Get authentication for the MCP server."""
|
||||
return None'''
|
||||
|
||||
return imports, get_auth
|
||||
|
||||
|
||||
def generate_tool_script(
|
||||
tool: mcp.types.Tool,
|
||||
server_url: str,
|
||||
auth_mode: str = "none",
|
||||
auth_value: str | None = None,
|
||||
) -> str:
|
||||
"""Generate a standalone Python script for an MCP tool.
|
||||
|
||||
Args:
|
||||
tool: MCP tool definition
|
||||
server_url: URL of the MCP server
|
||||
auth_mode: Authentication mode ("none", "oauth", "env_var", "token")
|
||||
auth_value: Auth value (env var name for env_var mode, token for token mode)
|
||||
|
||||
Returns:
|
||||
Complete Python script as a string
|
||||
"""
|
||||
function_name = to_snake_case(tool.name)
|
||||
|
||||
# Generate typed parameters
|
||||
params_str, param_names = generate_typed_params(tool.inputSchema)
|
||||
if not params_str:
|
||||
params_str = "" # No parameters
|
||||
|
||||
# Generate args dict
|
||||
args_dict = generate_args_dict(param_names)
|
||||
|
||||
# Generate auth code
|
||||
auth_imports, auth_function = generate_auth_code(auth_mode, auth_value, server_url)
|
||||
|
||||
imports = """import asyncio
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
from fastmcp import Client"""
|
||||
if auth_imports:
|
||||
imports += f"\n{auth_imports}"
|
||||
|
||||
return f'''\
|
||||
# /// script
|
||||
# dependencies = ["fastmcp>=2.0.0"]
|
||||
# ///
|
||||
|
||||
"""{tool.name}
|
||||
|
||||
{tool.description or ""}
|
||||
"""
|
||||
|
||||
{imports}
|
||||
|
||||
SERVER_URL = "{server_url}"
|
||||
|
||||
|
||||
{auth_function}
|
||||
|
||||
|
||||
async def {function_name}({params_str}) -> Any:
|
||||
"""{tool.description or tool.name}"""
|
||||
async with Client(SERVER_URL, auth=get_auth()) as client:
|
||||
result = await client.call_tool("{tool.name}", {args_dict})
|
||||
return result.data if result.data else result.content
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
params = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {{}}
|
||||
result = asyncio.run({function_name}(**params))
|
||||
print(json.dumps(result, indent=2) if not isinstance(result, str) else result)
|
||||
'''
|
||||
|
||||
|
||||
def generate_agents_md(
|
||||
server_name: str,
|
||||
server_url: str,
|
||||
tools: list[mcp.types.Tool],
|
||||
auth_mode: str = "none",
|
||||
auth_value: str | None = None,
|
||||
instructions: str | None = None,
|
||||
) -> str:
|
||||
"""Generate AGENTS.md documentation for agent usage.
|
||||
|
||||
Args:
|
||||
server_name: Name of the MCP server
|
||||
server_url: URL of the MCP server
|
||||
tools: List of MCP tools
|
||||
auth_mode: Authentication mode ("none", "oauth", "env_var", "token")
|
||||
auth_value: Auth value (env var name for env_var mode, token for token mode)
|
||||
instructions: Optional server-provided instructions for using the tools
|
||||
|
||||
Returns:
|
||||
Markdown documentation string
|
||||
"""
|
||||
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
|
||||
|
||||
# Build instructions section if provided
|
||||
instructions_section = ""
|
||||
if instructions:
|
||||
instructions_section = f"""
|
||||
## Server Instructions
|
||||
|
||||
{instructions}
|
||||
"""
|
||||
|
||||
# Generate tool list
|
||||
tool_lines = []
|
||||
for tool in tools:
|
||||
filename = to_snake_case(tool.name) + ".py"
|
||||
description = tool.description or tool.name
|
||||
tool_lines.append(f"- `{filename}` - {description}")
|
||||
|
||||
tools_list = "\n".join(tool_lines)
|
||||
|
||||
return f"""\
|
||||
# MCP Tools: {server_name}
|
||||
|
||||
Generated from: {server_url}
|
||||
Generated at: {timestamp}
|
||||
Tools: {len(tools)}
|
||||
{instructions_section}
|
||||
## Quick Start
|
||||
|
||||
```bash
|
||||
# Each script accepts JSON parameters
|
||||
uv run tool_name.py '{{"param1":"value1","param2":"value2"}}'
|
||||
|
||||
# No parameters (empty object)
|
||||
uv run tool_name.py '{{}}'
|
||||
|
||||
# Or omit for empty params
|
||||
uv run tool_name.py
|
||||
```
|
||||
|
||||
## Available Tools
|
||||
|
||||
{tools_list}
|
||||
|
||||
## Usage
|
||||
|
||||
Each script is standalone and can be:
|
||||
- **Run directly**: `uv run tool_name.py '{{"param":"value"}}'` (dependencies auto-installed)
|
||||
- **Imported**: `from tool_name import tool_name`
|
||||
- **Modified or deleted** without affecting other scripts
|
||||
|
||||
**Parameters:** All scripts accept JSON as first argument. Pass an empty object `{{}}` or omit for tools with no parameters.
|
||||
|
||||
**Dependencies:** Scripts use PEP 723 inline metadata, so `uv` automatically installs dependencies.
|
||||
|
||||
## Examples
|
||||
|
||||
```bash
|
||||
# Simple string parameter
|
||||
uv run greet.py '{{"name":"Alice"}}'
|
||||
|
||||
# Multiple parameters
|
||||
uv run greet.py '{{"name":"Alice","title":"Dr"}}'
|
||||
|
||||
# Complex nested objects
|
||||
uv run create_user.py '{{"profile":{{"name":"Alice","age":30}},"tags":["admin","user"]}}'
|
||||
|
||||
# No parameters
|
||||
uv run get_status.py
|
||||
```
|
||||
"""
|
||||
610
tests/cli/test_generate.py
Normal file
610
tests/cli/test_generate.py
Normal file
|
|
@ -0,0 +1,610 @@
|
|||
"""Tests for generate CLI command."""
|
||||
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import mcp.types
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.cli.cli import _infer_server_name
|
||||
from fastmcp.utilities.generate import (
|
||||
generate_agents_md,
|
||||
generate_args_dict,
|
||||
generate_auth_code,
|
||||
generate_tool_script,
|
||||
generate_typed_params,
|
||||
json_schema_to_python_type,
|
||||
to_snake_case,
|
||||
)
|
||||
|
||||
|
||||
class TestSnakeCase:
|
||||
def test_hyphenated_names(self):
|
||||
assert to_snake_case("get-document") == "get_document"
|
||||
assert to_snake_case("list-all-items") == "list_all_items"
|
||||
|
||||
def test_dotted_names(self):
|
||||
assert to_snake_case("chrome.getTabs") == "chrome_get_tabs"
|
||||
assert to_snake_case("list.items") == "list_items"
|
||||
|
||||
def test_camel_case(self):
|
||||
assert to_snake_case("getDocument") == "get_document"
|
||||
assert to_snake_case("listAllItems") == "list_all_items"
|
||||
|
||||
def test_mixed_formats(self):
|
||||
assert to_snake_case("get-documentId") == "get_document_id"
|
||||
assert to_snake_case("Chrome.getTabs") == "chrome_get_tabs"
|
||||
|
||||
|
||||
class TestJsonSchemaToType:
|
||||
def test_basic_types(self):
|
||||
assert json_schema_to_python_type({"type": "string"}) == "str"
|
||||
assert json_schema_to_python_type({"type": "integer"}) == "int"
|
||||
assert json_schema_to_python_type({"type": "number"}) == "float"
|
||||
assert json_schema_to_python_type({"type": "boolean"}) == "bool"
|
||||
assert json_schema_to_python_type({"type": "array"}) == "list"
|
||||
assert json_schema_to_python_type({"type": "object"}) == "dict"
|
||||
|
||||
def test_unknown_type(self):
|
||||
assert json_schema_to_python_type({"type": "unknown"}) == "Any"
|
||||
assert json_schema_to_python_type({}) == "Any"
|
||||
|
||||
|
||||
class TestTypedParams:
|
||||
def test_no_parameters(self):
|
||||
schema = {"type": "object", "properties": {}}
|
||||
params, names = generate_typed_params(schema)
|
||||
assert params == ""
|
||||
assert names == []
|
||||
|
||||
def test_required_parameter(self):
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
"required": ["name"],
|
||||
}
|
||||
params, names = generate_typed_params(schema)
|
||||
assert params == "name: str"
|
||||
assert names == ["name"]
|
||||
|
||||
def test_optional_parameter(self):
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {"age": {"type": "integer"}},
|
||||
}
|
||||
params, names = generate_typed_params(schema)
|
||||
assert params == "age: int | None = None"
|
||||
assert names == ["age"]
|
||||
|
||||
def test_mixed_parameters(self):
|
||||
schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"age": {"type": "integer"},
|
||||
"email": {"type": "string"},
|
||||
},
|
||||
"required": ["name", "email"],
|
||||
}
|
||||
params, names = generate_typed_params(schema)
|
||||
assert "name: str" in params
|
||||
assert "email: str" in params
|
||||
assert "age: int | None = None" in params
|
||||
assert names == ["name", "age", "email"]
|
||||
|
||||
|
||||
class TestArgsDict:
|
||||
def test_empty_params(self):
|
||||
result = generate_args_dict([])
|
||||
assert result == "{}"
|
||||
|
||||
def test_single_param(self):
|
||||
result = generate_args_dict(["name"])
|
||||
assert '"name": name' in result
|
||||
|
||||
def test_multiple_params(self):
|
||||
result = generate_args_dict(["name", "age", "email"])
|
||||
assert '"name": name' in result
|
||||
assert '"age": age' in result
|
||||
assert '"email": email' in result
|
||||
|
||||
|
||||
class TestAuthCode:
|
||||
def test_oauth_mode(self):
|
||||
imports, auth_func = generate_auth_code(
|
||||
"oauth", None, "https://example.com/mcp"
|
||||
)
|
||||
assert "from fastmcp.client.auth import OAuth" in imports
|
||||
assert 'OAuth(mcp_url="https://example.com/mcp")' in auth_func
|
||||
assert "def get_auth():" in auth_func
|
||||
|
||||
def test_env_var_mode(self):
|
||||
imports, auth_func = generate_auth_code(
|
||||
"env_var", "MY_API_TOKEN", "https://example.com/mcp"
|
||||
)
|
||||
assert imports == ""
|
||||
assert 'os.environ.get("MY_API_TOKEN")' in auth_func
|
||||
assert "Missing required environment variable: MY_API_TOKEN" in auth_func
|
||||
assert "def get_auth():" in auth_func
|
||||
|
||||
def test_token_mode(self):
|
||||
imports, auth_func = generate_auth_code(
|
||||
"token", "sk-test-123", "https://example.com/mcp"
|
||||
)
|
||||
assert imports == ""
|
||||
assert 'return "sk-test-123"' in auth_func
|
||||
assert "def get_auth():" in auth_func
|
||||
|
||||
def test_none_mode(self):
|
||||
imports, auth_func = generate_auth_code("none", None, "https://example.com/mcp")
|
||||
assert imports == ""
|
||||
assert "return None" in auth_func
|
||||
assert "def get_auth():" in auth_func
|
||||
|
||||
|
||||
class TestToolScript:
|
||||
def test_simple_tool(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="echo",
|
||||
description="Echo back the input",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {"text": {"type": "string"}},
|
||||
"required": ["text"],
|
||||
},
|
||||
)
|
||||
script = generate_tool_script(tool, "https://example.com/mcp")
|
||||
|
||||
# Check PEP 723 metadata
|
||||
assert "# /// script" in script
|
||||
assert '# dependencies = ["fastmcp>=2.0.0"]' in script
|
||||
|
||||
# Check function signature
|
||||
assert "async def echo(text: str) -> Any:" in script
|
||||
|
||||
# Check tool call
|
||||
assert 'await client.call_tool("echo"' in script
|
||||
|
||||
# Check server URL
|
||||
assert 'SERVER_URL = "https://example.com/mcp"' in script
|
||||
|
||||
# Check JSON parameter handling in __main__
|
||||
assert "import json" in script
|
||||
assert "import sys" in script
|
||||
assert "params = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}" in script
|
||||
assert "asyncio.run(echo(**params))" in script
|
||||
|
||||
def test_tool_with_no_params(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="get-status",
|
||||
description="Get server status",
|
||||
inputSchema={"type": "object", "properties": {}},
|
||||
)
|
||||
script = generate_tool_script(tool, "https://example.com/mcp")
|
||||
|
||||
# Should have function with no params
|
||||
assert "async def get_status() -> Any:" in script
|
||||
|
||||
def test_tool_with_optional_params(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="search",
|
||||
description="Search for items",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {"type": "string"},
|
||||
"limit": {"type": "integer"},
|
||||
},
|
||||
"required": ["query"],
|
||||
},
|
||||
)
|
||||
script = generate_tool_script(tool, "https://example.com/mcp")
|
||||
|
||||
# Check mixed params
|
||||
assert "query: str" in script
|
||||
assert "limit: int | None = None" in script
|
||||
|
||||
def test_tool_with_oauth_auth(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="echo",
|
||||
description="Echo back",
|
||||
inputSchema={"type": "object", "properties": {"text": {"type": "string"}}},
|
||||
)
|
||||
script = generate_tool_script(tool, "https://example.com/mcp", "oauth", None)
|
||||
|
||||
# Check OAuth import
|
||||
assert "from fastmcp.client.auth import OAuth" in script
|
||||
|
||||
# Check OAuth instantiation with mcp_url
|
||||
assert 'OAuth(mcp_url="https://example.com/mcp")' in script
|
||||
|
||||
def test_tool_with_env_var_auth(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="echo",
|
||||
description="Echo back",
|
||||
inputSchema={"type": "object", "properties": {"text": {"type": "string"}}},
|
||||
)
|
||||
script = generate_tool_script(
|
||||
tool, "https://example.com/mcp", "env_var", "MY_API_TOKEN"
|
||||
)
|
||||
|
||||
# Check env var reading
|
||||
assert 'os.environ.get("MY_API_TOKEN")' in script
|
||||
|
||||
# Check error handling
|
||||
assert "Missing required environment variable: MY_API_TOKEN" in script
|
||||
|
||||
# Should NOT have OAuth import
|
||||
assert "from fastmcp.client.auth import OAuth" not in script
|
||||
|
||||
def test_tool_with_token_auth(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="echo",
|
||||
description="Echo back",
|
||||
inputSchema={"type": "object", "properties": {"text": {"type": "string"}}},
|
||||
)
|
||||
script = generate_tool_script(
|
||||
tool, "https://example.com/mcp", "token", "sk-test-123"
|
||||
)
|
||||
|
||||
# Check embedded token
|
||||
assert 'return "sk-test-123"' in script
|
||||
|
||||
# Should NOT have OAuth import
|
||||
assert "from fastmcp.client.auth import OAuth" not in script
|
||||
|
||||
def test_tool_with_no_auth(self):
|
||||
tool = mcp.types.Tool(
|
||||
name="echo",
|
||||
description="Echo back",
|
||||
inputSchema={"type": "object", "properties": {"text": {"type": "string"}}},
|
||||
)
|
||||
script = generate_tool_script(tool, "https://example.com/mcp", "none", None)
|
||||
|
||||
# Check no auth
|
||||
assert "return None" in script
|
||||
|
||||
# Should NOT have OAuth import
|
||||
assert "from fastmcp.client.auth import OAuth" not in script
|
||||
|
||||
|
||||
class TestAgentsMd:
|
||||
def test_basic_generation_no_auth(self):
|
||||
tools = [
|
||||
mcp.types.Tool(
|
||||
name="echo",
|
||||
description="Echo back",
|
||||
inputSchema={"type": "object"},
|
||||
),
|
||||
mcp.types.Tool(
|
||||
name="reverse",
|
||||
description="Reverse text",
|
||||
inputSchema={"type": "object"},
|
||||
),
|
||||
]
|
||||
|
||||
md = generate_agents_md(
|
||||
"TestServer", "https://example.com", tools, "none", None, None
|
||||
)
|
||||
|
||||
assert "# MCP Tools: TestServer" in md
|
||||
assert "Generated from: https://example.com" in md
|
||||
assert "Tools: 2" in md
|
||||
assert "- `echo.py` - Echo back" in md
|
||||
assert "- `reverse.py` - Reverse text" in md
|
||||
# Check JSON parameter documentation
|
||||
assert "JSON parameters" in md or "JSON as first argument" in md
|
||||
assert "uv run" in md
|
||||
# Auth documentation should NOT be present
|
||||
assert "## Authentication" not in md
|
||||
|
||||
def test_no_auth_documentation_regardless_of_mode(self):
|
||||
"""Auth documentation should not be included in AGENTS.md regardless of auth mode."""
|
||||
tools = [
|
||||
mcp.types.Tool(
|
||||
name="test", description="Test", inputSchema={"type": "object"}
|
||||
),
|
||||
]
|
||||
|
||||
# Test all auth modes - none should generate auth documentation
|
||||
for auth_mode, auth_value in [
|
||||
("none", None),
|
||||
("oauth", None),
|
||||
("env_var", "MY_API_TOKEN"),
|
||||
("token", "sk-test"),
|
||||
]:
|
||||
md = generate_agents_md(
|
||||
"TestServer", "https://example.com", tools, auth_mode, auth_value, None
|
||||
)
|
||||
# No auth documentation should be present
|
||||
assert "## Authentication" not in md
|
||||
# Specific auth-related text should not be present
|
||||
if auth_mode == "env_var":
|
||||
assert "export MY_API_TOKEN" not in md
|
||||
assert "FASTMCP_AUTH_TOKEN" not in md
|
||||
|
||||
def test_server_instructions_included(self):
|
||||
tools = [
|
||||
mcp.types.Tool(
|
||||
name="test", description="Test", inputSchema={"type": "object"}
|
||||
),
|
||||
]
|
||||
instructions = (
|
||||
"Make sure to use these tools responsibly and follow rate limits."
|
||||
)
|
||||
md = generate_agents_md(
|
||||
"TestServer", "https://example.com", tools, "none", None, instructions
|
||||
)
|
||||
|
||||
assert "## Server Instructions" in md
|
||||
assert instructions in md
|
||||
|
||||
def test_no_instructions_section_when_none(self):
|
||||
tools = [
|
||||
mcp.types.Tool(
|
||||
name="test", description="Test", inputSchema={"type": "object"}
|
||||
),
|
||||
]
|
||||
md = generate_agents_md(
|
||||
"TestServer", "https://example.com", tools, "none", None, None
|
||||
)
|
||||
|
||||
assert "## Server Instructions" not in md
|
||||
|
||||
|
||||
class TestInferServerName:
|
||||
def test_simple_url(self):
|
||||
assert _infer_server_name("https://example.com/mcp") == "mcp"
|
||||
assert _infer_server_name("https://api.github.com/mcp") == "mcp"
|
||||
|
||||
def test_url_with_hostname(self):
|
||||
# Should remove www and common TLDs
|
||||
name = _infer_server_name("https://www.example.com")
|
||||
assert "example" in name
|
||||
|
||||
def test_complex_path(self):
|
||||
assert _infer_server_name("https://api.example.com/v1/mcp/tools") == "tools"
|
||||
|
||||
|
||||
class TestEndToEnd:
|
||||
async def test_generate_and_run_script(self):
|
||||
"""Test generating a script and executing it."""
|
||||
# Create a test server
|
||||
mcp = FastMCP("TestServer")
|
||||
|
||||
@mcp.tool
|
||||
def greet(name: str) -> str:
|
||||
"""Greet someone"""
|
||||
return f"Hello, {name}!"
|
||||
|
||||
# Connect and get tools
|
||||
async with Client(mcp) as client:
|
||||
tools = await client.list_tools()
|
||||
|
||||
# Generate script
|
||||
tool = tools[0]
|
||||
script = generate_tool_script(tool, "test://server")
|
||||
|
||||
# Write to temp file and verify it's valid Python
|
||||
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
|
||||
f.write(script)
|
||||
script_path = Path(f.name)
|
||||
|
||||
try:
|
||||
# Compile to check for syntax errors
|
||||
compile(script, str(script_path), "exec")
|
||||
finally:
|
||||
script_path.unlink()
|
||||
|
||||
async def test_full_generation_flow_with_instructions(self):
|
||||
"""Test complete flow: server with instructions -> files on disk."""
|
||||
# Create a test server with instructions
|
||||
mcp = FastMCP("TestServer")
|
||||
mcp.instructions = "Use these tools carefully. Rate limit: 100/min."
|
||||
|
||||
@mcp.tool
|
||||
def echo(text: str) -> str:
|
||||
"""Echo back text"""
|
||||
return text
|
||||
|
||||
@mcp.tool
|
||||
def reverse(text: str) -> str:
|
||||
"""Reverse text"""
|
||||
return text[::-1]
|
||||
|
||||
# Connect and get tools + server info
|
||||
async with Client(mcp) as client:
|
||||
tools = await client.list_tools()
|
||||
instructions = (
|
||||
client.initialize_result.instructions
|
||||
if client.initialize_result
|
||||
else None
|
||||
)
|
||||
server_name = (
|
||||
client.initialize_result.serverInfo.name
|
||||
if client.initialize_result and client.initialize_result.serverInfo
|
||||
else "test_server"
|
||||
)
|
||||
|
||||
# Generate files in temp directory
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
output_dir = Path(tmpdir) / "test_output"
|
||||
output_dir.mkdir()
|
||||
|
||||
# Generate tool scripts
|
||||
for tool in tools:
|
||||
script = generate_tool_script(tool, "test://server", "none", None)
|
||||
filename = to_snake_case(tool.name) + ".py"
|
||||
(output_dir / filename).write_text(script)
|
||||
|
||||
# Generate AGENTS.md
|
||||
agents_md = generate_agents_md(
|
||||
server_name, "test://server", tools, "none", None, instructions
|
||||
)
|
||||
(output_dir / "AGENTS.md").write_text(agents_md)
|
||||
|
||||
# Verify files exist
|
||||
assert (output_dir / "echo.py").exists()
|
||||
assert (output_dir / "reverse.py").exists()
|
||||
assert (output_dir / "AGENTS.md").exists()
|
||||
|
||||
# Verify AGENTS.md content
|
||||
agents_content = (output_dir / "AGENTS.md").read_text()
|
||||
assert "## Server Instructions" in agents_content
|
||||
assert "Use these tools carefully" in agents_content
|
||||
assert "Rate limit: 100/min" in agents_content
|
||||
assert "- `echo.py` - Echo back text" in agents_content
|
||||
assert "- `reverse.py` - Reverse text" in agents_content
|
||||
assert "## Authentication" not in agents_content # No auth docs
|
||||
|
||||
# Verify script content
|
||||
echo_script = (output_dir / "echo.py").read_text()
|
||||
assert "async def echo(text: str)" in echo_script
|
||||
assert 'SERVER_URL = "test://server"' in echo_script
|
||||
assert "return None" in echo_script # No auth mode
|
||||
|
||||
async def test_generation_with_different_auth_modes(self):
|
||||
"""Test generating scripts with all auth modes."""
|
||||
mcp = FastMCP("AuthTestServer")
|
||||
|
||||
@mcp.tool
|
||||
def test_tool() -> str:
|
||||
"""Test tool"""
|
||||
return "ok"
|
||||
|
||||
async with Client(mcp) as client:
|
||||
tools = await client.list_tools()
|
||||
|
||||
tool = tools[0]
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Test OAuth mode
|
||||
script_oauth = generate_tool_script(
|
||||
tool, "https://api.test.com", "oauth", None
|
||||
)
|
||||
oauth_path = Path(tmpdir) / "oauth.py"
|
||||
oauth_path.write_text(script_oauth)
|
||||
assert "from fastmcp.client.auth import OAuth" in script_oauth
|
||||
assert 'OAuth(mcp_url="https://api.test.com")' in script_oauth
|
||||
compile(script_oauth, str(oauth_path), "exec") # Verify syntax
|
||||
|
||||
# Test env var mode
|
||||
script_env = generate_tool_script(
|
||||
tool, "https://api.test.com", "env_var", "MY_TOKEN"
|
||||
)
|
||||
env_path = Path(tmpdir) / "env.py"
|
||||
env_path.write_text(script_env)
|
||||
assert 'os.environ.get("MY_TOKEN")' in script_env
|
||||
assert "Missing required environment variable: MY_TOKEN" in script_env
|
||||
compile(script_env, str(env_path), "exec") # Verify syntax
|
||||
|
||||
# Test token mode
|
||||
script_token = generate_tool_script(
|
||||
tool, "https://api.test.com", "token", "sk-test-123"
|
||||
)
|
||||
token_path = Path(tmpdir) / "token.py"
|
||||
token_path.write_text(script_token)
|
||||
assert 'return "sk-test-123"' in script_token
|
||||
compile(script_token, str(token_path), "exec") # Verify syntax
|
||||
|
||||
# Test none mode
|
||||
script_none = generate_tool_script(
|
||||
tool, "https://api.test.com", "none", None
|
||||
)
|
||||
none_path = Path(tmpdir) / "none.py"
|
||||
none_path.write_text(script_none)
|
||||
assert "return None" in script_none
|
||||
compile(script_none, str(none_path), "exec") # Verify syntax
|
||||
|
||||
|
||||
class TestCLICommand:
|
||||
"""Tests for the CLI command itself."""
|
||||
|
||||
async def test_auth_parsing_oauth(self):
|
||||
"""Test that --auth oauth is recognized."""
|
||||
auth_value = "oauth"
|
||||
# Simulate parsing logic
|
||||
if auth_value == "oauth":
|
||||
auth_mode = "oauth"
|
||||
elif auth_value.startswith("$"):
|
||||
auth_mode = "env_var"
|
||||
else:
|
||||
auth_mode = "token"
|
||||
assert auth_mode == "oauth"
|
||||
|
||||
async def test_auth_parsing_env_var(self):
|
||||
"""Test that --auth $VAR is recognized as env var."""
|
||||
import os
|
||||
|
||||
# Set up an env var
|
||||
os.environ["TEST_TOKEN"] = "test-value"
|
||||
|
||||
auth_value = "$TEST_TOKEN"
|
||||
# Simulate parsing logic
|
||||
if auth_value == "oauth":
|
||||
auth_mode = "oauth"
|
||||
elif auth_value.startswith("$"):
|
||||
auth_mode = "env_var"
|
||||
env_var_name = auth_value[1:]
|
||||
token = os.environ.get(env_var_name)
|
||||
assert token == "test-value"
|
||||
else:
|
||||
auth_mode = "token"
|
||||
assert auth_mode == "env_var"
|
||||
|
||||
del os.environ["TEST_TOKEN"]
|
||||
|
||||
async def test_auth_parsing_literal_token(self):
|
||||
"""Test that --auth with literal value is treated as token."""
|
||||
auth_value = "sk-test-123"
|
||||
# Simulate parsing logic
|
||||
if auth_value == "oauth":
|
||||
auth_mode = "oauth"
|
||||
elif auth_value.startswith("$"):
|
||||
auth_mode = "env_var"
|
||||
else:
|
||||
auth_mode = "token"
|
||||
assert auth_mode == "token"
|
||||
assert auth_value == "sk-test-123"
|
||||
|
||||
async def test_cli_missing_env_var_fails(self):
|
||||
"""Test that missing env var when using $VAR would fail."""
|
||||
import os
|
||||
|
||||
# Ensure env var doesn't exist
|
||||
if "NONEXISTENT_VAR" in os.environ:
|
||||
del os.environ["NONEXISTENT_VAR"]
|
||||
|
||||
# Simulate the CLI check
|
||||
auth_value = "$NONEXISTENT_VAR"
|
||||
env_var_name = auth_value[1:]
|
||||
token = os.environ.get(env_var_name)
|
||||
assert token is None # Would cause CLI to exit with error
|
||||
|
||||
async def test_generated_scripts_are_executable(self):
|
||||
"""Test that generated scripts can actually be imported and used."""
|
||||
mcp = FastMCP("ExecutableTest")
|
||||
|
||||
@mcp.tool
|
||||
def multiply(a: int, b: int) -> int:
|
||||
"""Multiply two numbers"""
|
||||
return a * b
|
||||
|
||||
async with Client(mcp) as client:
|
||||
tools = await client.list_tools()
|
||||
|
||||
tool = tools[0]
|
||||
script = generate_tool_script(tool, "test://server", "none", None)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
script_path = Path(tmpdir) / "multiply.py"
|
||||
script_path.write_text(script)
|
||||
|
||||
# Verify it's valid Python
|
||||
compile(script, str(script_path), "exec")
|
||||
|
||||
# Verify key components are present
|
||||
assert "async def multiply(a: int, b: int) -> Any:" in script
|
||||
assert 'await client.call_tool("multiply"' in script
|
||||
assert '"a": a' in script
|
||||
assert '"b": b' in script
|
||||
Loading…
Add table
Add a link
Reference in a new issue