Newer
Older
navi-1 / mcp-servers / _template / app / mcp_server.py
"""MCP server template — heavily commented reference for Navi.

Copy this file into your new server at:
    mcp-servers/<your_name>/app/mcp_server.py

Then edit it, adding 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

# ── 1. INSTRUCTIONS ──────────────────────────────────────────────────────
# These instructions are sent to Navi during the MCP handshake.
# They become part of Navi's system prompt when this server is enabled
# in a profile. Write them carefully — they tell Navi WHEN and HOW to
# use your tools. Include workflow, rules, and an ABSOLUTE RULE about
# never bypassing MCP with filesystem/terminal.
# ────────────────────────────────────────────────────────────────────────
INSTRUCTIONS = """
Replace this with instructions for Navi.

Example:
    MyServer MCP server provides X and Y tools.

    Use it when the task involves:
    - doing something that only this server can do;
    - ...

    Workflow:
    1. tool_a — do step one.
    2. tool_b — do step two.

    ABSOLUTE RULE — NEVER bypass MCP tools:
    You MUST NOT use filesystem, terminal, code_exec, or any direct
    file access for operations covered by this server.
""".strip()


# ── 2. FastMCP INSTANCE ─────────────────────────────────────────────────
# Create the server. The name should match the directory name under
# mcp-servers/ and the entry in mcp_servers.json.
mcp = FastMCP("template", instructions=INSTRUCTIONS)


# ── 3. HELPER ─────────────────────────────────────────────────────────
# A small helper to return JSON strings from tools. MCP tools return
# plain text — JSON is a good convention for structured data.
def _json(data: Any) -> str:
    return json.dumps(data, ensure_ascii=False, indent=2)


# ── 4. TOOL DEFINITIONS ────────────────────────────────────────────────
# EVERY @mcp.tool MUST be defined HERE, BEFORE the main() block below.
# mcp.run() in main() enters an infinite stdio loop — tools defined AFTER
# that line are NEVER registered and will be invisible to Navi.
#
# Rules:
#   - async def only
#   - return str (plain text or JSON)
#   - raise Exception on real errors (FastMCP will catch and report)
#   - use .get() for optional params; validate required params explicitly
#   - Parameters MUST use Annotated[..., Field(description=...)]
# ────────────────────────────────────────────────────────────────────────

@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."""
    result = a + b
    return _json({"a": a, "b": b, "result": result})


# ── 5. MAIN / TRANSPORT ────────────────────────────────────────────────
# ⚠ CRITICAL: Do NOT define any @mcp.tool functions after this line.
# mcp.run() blocks forever — tools placed below are NEVER registered.
#
# The server supports stdio (default), sse, and streamable-http.
# Navi always connects via stdio — so keep transport="stdio" as default.
# The TRANSPORT env var lets you test other transports manually.
def main() -> None:
    transport = os.environ.get("MCP_TRANSPORT", "stdio")
    if transport not in {"stdio", "sse", "streamable-http"}:
        raise SystemExit("MCP_TRANSPORT must be stdio, sse, or streamable-http")
    mcp.run(transport=transport)  # type: ignore[arg-type]


if __name__ == "__main__":
    main()