Compare commits

...

1 commit

Author SHA1 Message Date
jake@prefect.io
fb28f9cda9 Add fastmcp generate command for creating standalone tool scripts
Generates Python scripts from MCP server tools for progressive discovery by agents.

Usage:
  # No auth
  fastmcp generate https://api.example.com/mcp

  # OAuth
  fastmcp generate https://api.example.com/mcp --auth oauth

  # Environment variable
  fastmcp generate https://api.example.com/mcp --auth '$MY_TOKEN'

  # Embedded token
  fastmcp generate https://api.example.com/mcp --auth "sk-token-123"

Each tool becomes a standalone script with inline dependencies (PEP 723).
Server instructions are included in AGENTS.md when provided.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-11-05 17:33:39 -05:00
3 changed files with 1108 additions and 0 deletions

View file

@ -10,6 +10,7 @@ import sys
from contextlib import contextmanager from contextlib import contextmanager
from pathlib import Path from pathlib import Path
from typing import Annotated, Literal from typing import Annotated, Literal
from urllib.parse import urlparse
import cyclopts import cyclopts
import pyperclip import pyperclip
@ -17,10 +18,17 @@ from rich.console import Console
from rich.table import Table from rich.table import Table
import fastmcp import fastmcp
from fastmcp import Client
from fastmcp.cli import run as run_module from fastmcp.cli import run as run_module
from fastmcp.cli.install import install_app from fastmcp.cli.install import install_app
from fastmcp.client.auth import OAuth
from fastmcp.server.server import FastMCP from fastmcp.server.server import FastMCP
from fastmcp.utilities.cli import is_already_in_uv_subprocess, load_and_merge_config 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 ( from fastmcp.utilities.inspect import (
InspectFormat, InspectFormat,
format_info, format_info,
@ -867,6 +875,186 @@ async def prepare(
sys.exit(1) 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 # Add project subcommand group
app.command(project_app) app.command(project_app)

View 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
View 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