mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-24 06:24:18 +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 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)
|
||||||
|
|
||||||
|
|
|
||||||
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