"""Built-in tool that scaffolds a new MCP server directory."""
import asyncio
import json
import shutil
import subprocess
from pathlib import Path
from navi.config import settings
from ._internal.base import Tool, ToolContext, ToolResult
# Template for pyproject.toml — placeholders {name}, {description}, {deps}
_PYPROJECT_TEMPLATE = """[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[project]
name = "mcp-server-{name}"
version = "0.1.0"
description = "{description}"
requires-python = ">=3.11"
dependencies = [
"mcp>=1.27",
"pydantic>=2.0",
{deps}
]
[project.scripts]
mcp-server-{name} = "app.mcp_server:main"
[tool.setuptools.packages.find]
where = ["."]
include = ["app*"]
"""
_README_TEMPLATE = """# MCP Server: {name}
{description}
## Tools
TODO: list tools and their purposes.
## Setup
```bash
python -m venv .venv
source .venv/bin/activate
pip install -e .
```
## Navi registration
Create a file `mcp_servers.d/{name}.json` with the server config:
- transport: stdio
- command: absolute path to `.venv/bin/python`
- args: `["-m", "app.mcp_server"]`
- cwd: absolute path to this directory
- groups: map tool names to logical groups
"""
class CreateMcpServerTool(Tool):
name = "create_mcp_server"
description = (
"Scaffold a new MCP server directory under mcp-servers/<name>/ with "
"pyproject.toml, app/mcp_server.py (from the template with inline comments), "
"README.md, and a virtual environment. Then installs dependencies. "
"Optionally auto-registers the server by creating mcp_servers.d/<name>.json. "
"After this returns, you must still edit app/mcp_server.py and call reload_tools."
)
parameters = {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Server directory name (snake_case or kebab-case). Will become mcp-servers/<name>/.",
},
"description": {
"type": "string",
"description": "Short description for pyproject.toml and README.",
},
"dependencies": {
"type": "array",
"items": {"type": "string"},
"description": "Extra pip dependencies (e.g. httpx, asyncpg). mcp and pydantic are added automatically.",
},
"auto_register": {
"type": "boolean",
"description": "If true, create mcp_servers.d/<name>.json automatically so no manual filesystem edit is needed.",
},
},
"required": ["name", "description"],
}
async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult:
name = (params.get("name") or "").strip()
description = params.get("description", "")
dependencies: list[str] = params.get("dependencies") or []
auto_register: bool = params.get("auto_register", False)
if not name:
return ToolResult(success=False, output="name is required.", error="missing name")
if name.startswith("_"):
return ToolResult(success=False, output="name must not start with '_'.", error="invalid name")
base_dir = Path("mcp-servers") / name
if base_dir.exists():
return ToolResult(
success=False,
output=f"Directory already exists: {base_dir}",
error="already exists",
)
# 1. Create directories
app_dir = base_dir / "app"
app_dir.mkdir(parents=True)
# 2. Write pyproject.toml
deps_lines = ""
if dependencies:
deps_lines = "\n".join(f' "{d}",' for d in dependencies)
pyproject = _PYPROJECT_TEMPLATE.format(
name=name,
description=description.replace('"', '\\"'),
deps=deps_lines,
)
(base_dir / "pyproject.toml").write_text(pyproject, encoding="utf-8")
# 3. Copy template mcp_server.py
template_server = Path("mcp-servers") / "_template" / "app" / "mcp_server.py"
if template_server.exists():
server_code = template_server.read_text(encoding="utf-8")
else:
server_code = _FALLBACK_SERVER_TEMPLATE
(app_dir / "mcp_server.py").write_text(server_code, encoding="utf-8")
# 4. __init__.py
(app_dir / "__init__.py").write_text("", encoding="utf-8")
# 5. README
readme = _README_TEMPLATE.format(name=name, description=description)
(base_dir / "README.md").write_text(readme, encoding="utf-8")
# 6. Create venv and install (off-loaded to a thread to avoid any
# event-loop/subprocess interaction issues under uvicorn/anyio).
abs_dir = base_dir.resolve()
venv_dir = abs_dir / ".venv"
python_bin = venv_dir / "bin" / "python"
pip_bin = venv_dir / "bin" / "pip"
def _setup() -> tuple[bool, str]:
"""Return (ok, error_msg). Runs in a worker thread."""
# --- Create venv ---
result = subprocess.run(
["python", "-m", "venv", str(venv_dir)],
capture_output=True,
text=True,
timeout=30.0,
)
if result.returncode != 0:
return False, f"venv creation failed:\n{result.stderr}"
# --- Install deps ---
install_args = [
str(pip_bin), "install", "-e", str(base_dir),
"mcp>=1.27", "pydantic>=2.0",
]
for dep in dependencies:
install_args.append(dep)
result = subprocess.run(
install_args,
capture_output=True,
text=True,
timeout=120.0,
)
if result.returncode != 0:
return False, f"pip install failed:\n{result.stderr}"
return True, ""
try:
ok, err_msg = await asyncio.wait_for(
asyncio.to_thread(_setup),
timeout=150.0,
)
if not ok:
return ToolResult(
success=False,
output=err_msg,
error="setup failed",
)
except asyncio.TimeoutError:
return ToolResult(
success=False,
output="Setup timed out (venv or pip install took too long).",
error="timeout",
)
except Exception as exc:
return ToolResult(
success=False,
output=f"Setup error: {exc}",
error="setup error",
)
# 7. Verify syntax
server_py_path = app_dir / "mcp_server.py"
try:
import py_compile
py_compile.compile(str(server_py_path), doraise=True)
except Exception as exc:
return ToolResult(
success=False,
output=f"Syntax check failed (unexpected — template should be valid): {exc}",
error="syntax error",
)
# 8. Optional auto-register in mcp_servers.d/
register_note = ""
if auto_register:
from navi.mcp.config import McpServerConfig, save_mcp_servers, load_mcp_servers
try:
configs = load_mcp_servers()
configs[name] = McpServerConfig(
transport="stdio",
command=str(python_bin),
args=["-m", "app.mcp_server"],
cwd=str(abs_dir),
env={"MCP_TRANSPORT": "stdio"},
groups={"default": []},
)
save_mcp_servers(configs)
register_note = (
f"\nAuto-registered in mcp_servers.d/{name}.json.\n"
f"Call reload_tools to connect.\n"
)
except Exception as exc:
register_note = (
f"\nWARNING: auto_register failed: {exc}\n"
f"You must manually create mcp_servers.d/{name}.json.\n"
)
# Return connection instructions
output = (
f"Created MCP server at: {abs_dir}\n\n"
f"Next steps:\n"
f"1. Edit {abs_dir / 'app' / 'mcp_server.py'} — add your tools and INSTRUCTIONS.\n"
)
if not auto_register:
output += (
f"2. Create mcp_servers.d/{name}.json with:\n"
f'{{\n'
f' "transport": "stdio",\n'
f' "command": "{python_bin}",\n'
f' "args": ["-m", "app.mcp_server"],\n'
f' "cwd": "{abs_dir}",\n'
f' "env": {{"MCP_TRANSPORT": "stdio"}},\n'
f' "groups": {{\n'
f' "default": []\n'
f' }}\n'
f'}}\n'
f"3. Call reload_tools to connect.\n"
f"4. Call test_mcp_tool to verify each tool.\n"
)
else:
output += register_note + "2. Call test_mcp_tool to verify each tool.\n"
return ToolResult(success=True, output=output)
_FALLBACK_SERVER_TEMPLATE = '''"""MCP server — replace with your own tools and instructions."""
from __future__ import annotations
import json
import os
from typing import Annotated, Any
from mcp.server.fastmcp import FastMCP
from pydantic import Field
INSTRUCTIONS = """
Replace this with instructions for Navi.
Describe what this server does, when to use it, and the workflow.
Add an ABSOLUTE RULE about never bypassing these tools.
""".strip()
mcp = FastMCP("server", instructions=INSTRUCTIONS)
def _json(data: Any) -> str:
return json.dumps(data, ensure_ascii=False, indent=2)
@mcp.tool(name="hello")
async def hello_tool(
name: Annotated[str, Field(description="Name to greet.")],
) -> str:
"""Say hello to someone."""
return f"Hello, {name}!"
@mcp.tool(name="add")
async def add_tool(
a: Annotated[int, Field(description="First number.")],
b: Annotated[int, Field(description="Second number.")],
) -> str:
"""Add two numbers."""
return _json({"a": a, "b": b, "result": a + b})
def main() -> None:
transport = os.environ.get("MCP_TRANSPORT", "stdio")
mcp.run(transport=transport)
if __name__ == "__main__":
main()
'''
