diff --git a/src/fastmcp/cli/cli.py b/src/fastmcp/cli/cli.py index 17d7e1c99..f34ca8947 100644 --- a/src/fastmcp/cli/cli.py +++ b/src/fastmcp/cli/cli.py @@ -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) diff --git a/src/fastmcp/utilities/generate.py b/src/fastmcp/utilities/generate.py new file mode 100644 index 000000000..3348e851b --- /dev/null +++ b/src/fastmcp/utilities/generate.py @@ -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 +``` +""" diff --git a/tests/cli/test_generate.py b/tests/cli/test_generate.py new file mode 100644 index 000000000..95b78bb8b --- /dev/null +++ b/tests/cli/test_generate.py @@ -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