diff --git a/manuals/write_mcp_server.md b/manuals/write_mcp_server.md new file mode 100644 index 0000000..a0e8ec0 --- /dev/null +++ b/manuals/write_mcp_server.md @@ -0,0 +1,224 @@ +# Writing MCP Servers for Navi + +This manual describes how to create, test, register, and maintain MCP servers that extend Navi's capabilities. Read this **before** you start building a new server. + +## 1. Philosophy: Why MCP instead of user tools? + +MCP servers run in isolated processes and communicate via the Model Context Protocol. They **cannot** crash Navi's core, they can be reloaded without restarting the server, and they scale to complex external integrations (APIs, databases, browsers, etc.). The trade-off is slightly more boilerplate than a single `tools/foo.py` file. + +**Rule:** Every new capability that is not trivial (more than a simple datetime or notes lookup) should be built as an MCP server. + +## 2. Directory structure + +MCP servers live under: + +``` +mcp-servers// +├── pyproject.toml +├── README.md +└── app/ + ├── __init__.py + └── mcp_server.py +``` + +- `` — snake_case or kebab-case. Must match the key you will use in `mcp_servers.d/.json`. +- `pyproject.toml` — Python package metadata and dependencies. +- `app/mcp_server.py` — the actual server code (FastMCP). + +## 3. Creating a new server from the template + +Use the built-in tool `create_mcp_server` (preferred) or copy the template manually: + +```bash +cp -r mcp-servers/_template mcp-servers/my_server +cd mcp-servers/my_server +# Edit pyproject.toml: change name, description, add dependencies +# Edit app/mcp_server.py: add your tools and instructions +``` + +### 3.1 pyproject.toml + +Minimal required fields: + +```toml +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "mcp-server-myserver" +version = "0.1.0" +description = "What this server does" +requires-python = ">=3.11" +dependencies = [ + "mcp>=1.27", + "pydantic>=2.0", + # add your own: httpx, asyncpg, playwright, etc. +] + +[project.scripts] +mcp-server-myserver = "app.mcp_server:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["app*"] +``` + +### 3.2 app/mcp_server.py + +Read the template at `mcp-servers/_template/app/mcp_server.py` first. It contains a working hello-world server with extensive inline comments. + +Key sections you must edit: + +1. **`INSTRUCTIONS`** — These are injected into Navi's system prompt. Describe: + - What this server does and when to use it. + - Recommended workflow (order of tool calls). + - ABSOLUTE RULE about never bypassing these tools with filesystem/terminal. + +2. **`mcp = FastMCP("name", instructions=INSTRUCTIONS)`** — The name should match the directory key. + +3. **Tool functions** — Each tool: + - Is an `async def`. + - Uses `@mcp.tool(name="tool_name")`. + - Parameters use `Annotated[T, Field(description="...")]` — never plain types. + - Returns a plain `str` (JSON string for structured data is fine). + - Raises on real errors. + - Validates required params explicitly. + +Example: + +```python +@mcp.tool(name="search_docs") +async def search_docs_tool( + query: Annotated[str, Field(description="Search query string.")], + limit: Annotated[int, Field(description="Max results.")] = 10, +) -> str: + """Search the documentation index.""" + if not query.strip(): + raise ValueError("query is required and cannot be empty.") + results = await _do_search(query, limit) + return json.dumps(results, ensure_ascii=False, indent=2) +``` + +## 4. Environment and installation + +After creating files, you **must**: + +1. Create a virtual environment: + ```bash + python -m venv .venv + source .venv/bin/activate + pip install -e . + ``` + +2. Verify the server starts without crashing: + ```bash + timeout 5 python -m app.mcp_server || true + ``` + If it prints a traceback, fix the code before proceeding. + +## 5. Registering the server in Navi + +Create a file `mcp_servers.d/.json` in the project root. The filename (without `.json`) becomes the server name. Example for a server named `my_server`: + +```json +{ + "transport": "stdio", + "command": "/absolute/path/to/mcp-servers/my_server/.venv/bin/python", + "args": ["-m", "app.mcp_server"], + "cwd": "/absolute/path/to/mcp-servers/my_server", + "env": { + "MCP_TRANSPORT": "stdio" + }, + "groups": { + "default": ["search_docs", "read_doc"] + }, + "instructions": "Optional extra instructions merged with the server's own INSTRUCTIONS." +} +``` + +**Critical fields:** +- `command` — absolute path to the venv's Python binary. +- `cwd` — absolute path to the server directory. +- The filename must be `.json` (e.g. `my_server.json`). +- `args` — usually `["-m", "app.mcp_server"]`. +- `groups` — organize tools into named groups so profiles can reference them cleanly. + +After editing `mcp_servers.d/.json`, call `reload_tools` to connect the server and register its tools. + +## 6. Testing + +### 6.1 Check connection + +Call `mcp_status`. You should see your server as `connected` with the correct tool count. + +### 6.2 Test each tool + +Call `test_mcp_tool` for every tool your server exposes: + +``` +test_mcp_tool(server_name="my_server", tool_name="search_docs", arguments={"query": "hello", "limit": 3}) +``` + +If any tool fails, read the error output, fix the code in `app/mcp_server.py`, and repeat. + +### 6.3 Manual stderr inspection + +If `mcp_status` shows `disconnected` but the code looks correct, inspect stderr manually: + +```bash +cd mcp-servers/my_server +.venv/bin/python -m app.mcp_server 2>&1 | head -n 20 +``` + +## 7. Updating an MCP server + +1. Edit the code in `mcp-servers//app/mcp_server.py`. +2. (Optional) If you added new dependencies, edit `pyproject.toml` and run `pip install -e .` inside the venv. +3. Call `reload_tools` to reconnect the server and re-register tools. +4. Call `test_mcp_tool` to verify. + +## 8. Deleting an MCP server + +1. Remove the server directory or move it to a backup location. +2. Remove the entry from `mcp_servers.d/.json`. +3. Call `reload_tools`. + +## 9. Connecting an external MCP server + +If the server was written by someone else: + +1. Clone or place the server code on disk. +2. Create its venv and install dependencies. +3. Read its README to learn tool names and required environment variables. +4. Add an entry to `mcp_servers.d/.json` with the correct `command`, `cwd`, `args`, and `env`. +5. Define `groups` mapping the tools into logical sets. +6. Call `reload_tools`. +7. Call `test_mcp_tool` for a representative tool. + +## 10. Common mistakes and debugging + +| Symptom | Cause | Fix | +|---|---|---| +| `mcp_status` shows `disconnected` | Wrong `command` or `cwd` path | Double-check absolute paths | +| Traceback on startup | Syntax error or missing import | Run `python -m py_compile app/mcp_server.py` | +| `test_mcp_tool` returns `is_error=True` | Tool raised an exception | Fix the tool logic; check parameter validation | +| Tool schema missing descriptions | Used plain types instead of `Annotated[..., Field(...)]` | Add `Field(description=...)` to every parameter | +| Navi never calls the server | Profile does not map the server in `mcp_servers` | Edit the profile's `config.json` and add the server groups | +| Navi bypasses MCP with filesystem | `INSTRUCTIONS` missing ABSOLUTE RULE | Add explicit rule in server INSTRUCTIONS | + +## 11. Workflow checklist for Navi + +When asked to create a new MCP server: + +1. Read this manual (`manuals/write_mcp_server.md`). +2. Read the template (`mcp-servers/_template/app/mcp_server.py`). +3. Call `create_mcp_server(name=..., description=...)` to scaffold the directory. +4. Edit `app/mcp_server.py` iteratively using `filesystem`. +5. Validate syntax: `code_exec` or `terminal` with `python -m py_compile ...`. +6. Test startup: `terminal` with `timeout 5 python -m app.mcp_server`. +7. Edit `mcp_servers.d/.json` via `filesystem` to register the server. +8. Call `reload_tools`. +9. Call `test_mcp_tool` for every tool. +10. Write good `INSTRUCTIONS` inside `mcp_server.py`. +11. Report results to the user. diff --git a/mcp-servers/_template/README.md b/mcp-servers/_template/README.md new file mode 100644 index 0000000..b6dfa3f --- /dev/null +++ b/mcp-servers/_template/README.md @@ -0,0 +1,2 @@ +# MCP Server Template +## Замени это описание на описание своего сервера diff --git a/mcp-servers/_template/app/__init__.py b/mcp-servers/_template/app/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/mcp-servers/_template/app/__init__.py diff --git a/mcp-servers/_template/app/mcp_server.py b/mcp-servers/_template/app/mcp_server.py new file mode 100644 index 0000000..0afdd02 --- /dev/null +++ b/mcp-servers/_template/app/mcp_server.py @@ -0,0 +1,104 @@ +"""MCP server template — heavily commented reference for Navi. + +Copy this file into your new server at: + mcp-servers//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() diff --git a/mcp-servers/_template/pyproject.toml b/mcp-servers/_template/pyproject.toml new file mode 100644 index 0000000..56b9cdc --- /dev/null +++ b/mcp-servers/_template/pyproject.toml @@ -0,0 +1,20 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "mcp-server-template" +version = "0.1.0" +description = "MCP server template — replace with your server name and description" +requires-python = ">=3.11" +dependencies = [ + "mcp>=1.27", + "pydantic>=2.0", +] + +[project.scripts] +mcp-server-template = "app.mcp_server:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["app*"] diff --git a/mcp-servers/project_health/README.md b/mcp-servers/project_health/README.md new file mode 100644 index 0000000..f9c6237 --- /dev/null +++ b/mcp-servers/project_health/README.md @@ -0,0 +1,24 @@ +# MCP Server: project_health + +Analyzes codebase health, including statistics, markers (TODO/FIXME), secrets detection, duplicate files, and dependency analysis. + +## 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/project_health.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 diff --git a/mcp-servers/project_health/app/__init__.py b/mcp-servers/project_health/app/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/mcp-servers/project_health/app/__init__.py diff --git a/mcp-servers/project_health/app/mcp_server.py b/mcp-servers/project_health/app/mcp_server.py new file mode 100644 index 0000000..fb64e07 --- /dev/null +++ b/mcp-servers/project_health/app/mcp_server.py @@ -0,0 +1,233 @@ +"""MCP server for project_health — Analyzes project structure, finds duplicates, and detects dependencies.""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +from pathlib import Path +from typing import Annotated, Any + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +INSTRUCTIONS = """ +project_health provides tools to analyze the health and structure of a codebase. + +Use it when the task involves: +- summarizing project statistics (files, lines, languages). +- finding TODO/FIXME markers or potential secrets in the codebase. +- identifying duplicate files based on content. +- detecting project dependencies from configuration files. + +Workflow: +1. get_project_summary — check the overall state and potential issues. +2. find_duplicate_files — clean up redundant files. +3. get_project_dependencies — understand the project's external requirements. + +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() + +mcp = FastMCP("project_health", instructions=INSTRUCTIONS) + + +def _json(data: Any) -> str: + return json.dumps(data, ensure_ascii=False, indent=2) + + +# ── TOOL DEFINITIONS ──────────────────────────────────────── +# ALL @mcp.tool decorators MUST be placed here, BEFORE main(). + +@mcp.tool(name="get_project_summary") +async def get_project_summary( + path: Annotated[str, Field(description="Absolute path to the project root.")], +) -> str: + """Summarize project stats, markers (TODO/FIXME), and potential secrets.""" + root = Path(path) + if not root.is_dir(): + return _json({"error": f"Path {path} is not a directory."}) + + stats = {"total_files": 0, "total_lines": 0, "languages": {}} + markers = [] + secrets_found = [] + + # Patterns for secrets + secret_patterns = { + "API Key": re.compile(r"(?i)(api[_-]?key|token|secret|password|auth)[\\s:=]+['\"][a-zA-Z0-9]{16,}[\'\"]"), + "Generic Secret": re.compile(r"(?i)password\s*=\s*['\"][^'\"]+['\"]"), + } + + exclude_dirs = {".git", "node_modules", "__pycache__", ".venv", "venv", ".pytest_cache", "dist", "build"} + + for dirpath, dirnames, filenames in os.walk(root): + # Prune excluded directories + dirnames[:] = [d for d in dirnames if d not in exclude_dirs] + + for filename in filenames: + file_path = Path(dirpath) / filename + try: + # Skip binary files or very large files for summary + if file_path.stat().st_size > 1_000_000: # 1MB limit for scanning + continue + + stats["total_files"] += 1 + + # Determine language by extension + ext = file_path.suffix.lower() + if ext in ['.py']: lang = 'Python' + elif ext in ['.js', '.ts']: lang = 'JavaScript/TypeScript' + elif ext in ['.md']: lang = 'Markdown' + elif ext in ['.json']: lang = 'JSON' + elif ext in ['.toml']: lang = 'TOML' + elif ext in ['.c', '.cpp', '.h']: lang = 'C/C++' + else: lang = 'Other' + + stats["languages"][lang] = stats["languages"].get(lang, 0) + 1 + + # Read content for markers and secrets + with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: + lines = f.readlines() + stats["total_lines"] += len(lines) + + for i, line in enumerate(lines, 1): + # Check for TODO/FIXME + if "TODO" in line or "FIXME" in line: + markers.append({ + "file": str(file_path.relative_to(root)), + "line": i, + "content": line.strip() + }) + + # Check for secrets + for name, pattern in secret_patterns.items(): + if pattern.search(line): + secrets_found.append({ + "file": str(file_path.relative_to(root)), + "line": i, + "type": name + }) + except Exception: + continue + + return _json({ + "file_stats": stats, + "markers": markers, + "secrets_found": secrets_found + }) + + +@mcp.tool(name="find_duplicate_files") +async def find_duplicate_files( + path: Annotated[str, Field(description="Absolute path to the project root.")], +) -> str: + """Find files with identical content using SHA256 hashing.""" + root = Path(path) + if not root.is_dir(): + return _json({"error": f"Path {path} is not a directory."}) + + hashes = {} # hash -> [list of paths] + exclude_dirs = {".git", "node_modules", "__pycache__", ".venv", "venv"} + + for dirpath, dirnames, filenames in os.walk(root): + dirnames[:] = [d for d in dirnames if d not in exclude_dirs] + + for filename in filenames: + file_path = Path(dirpath) / filename + try: + # Only hash files up to 5MB to avoid performance issues + if file_path.stat().st_size > 5_000_000: + continue + + hasher = hashlib.sha256() + with open(file_path, 'rb') as f: + while chunk := f.read(8192): + hasher.update(chunk) + + file_hash = hasher.hexdigest() + rel_path = str(file_path.relative_to(root)) + + if file_hash in hashes: + hashes[file_hash].append(rel_path) + else: + hashes[file_hash] = [rel_path] + except Exception: + continue + + # Filter only groups that have more than one file + duplicates = [paths for paths in hashes.values() if len(paths) > 1] + return _json({"duplicate_groups": duplicates}) + + +@mcp.tool(name="get_project_dependencies") +async def get_project_dependencies( + path: Annotated[str, Field(description="Absolute path to the project root.")], +) -> str: + """Identify dependencies by parsing common configuration files.""" + root = Path(path) + if not root.is_dir(): + return _json({"error": f"Path {path} is not a directory."}) + + dependencies = { + "python": [], + "javascript": [], + "other": [] + } + + # Check pyproject.toml + pyproject = root / "pyproject.toml" + if pyproject.exists(): + try: + content = pyproject.read_text(encoding='utf-8') + # Simple regex to find dependencies in pyproject.toml + deps = re.findall(r'dependencies\s*=\s*\[(.*?)\]', content, re.DOTALL) + if deps: + # Clean up the matches + dep_list = [d.strip().strip('"').strip("'") for d in re.split(r',', deps[0])] + dependencies["python"].extend([d for d in dep_list if d]) + except Exception: + pass + + # Check requirements.txt + req_txt = root / "requirements.txt" + if req_txt.exists(): + try: + content = req_txt.read_text(encoding='utf-8') + deps = [line.strip() for line in content.splitlines() if line.strip() and not line.startswith("#")] + dependencies["python"].extend(deps) + except Exception: + pass + + # Check package.json + package_json = root / "package.json" + if package_json.exists(): + try: + data = json.loads(package_json.read_text(encoding='utf-8')) # Note: error handling needed + # This is a simplified parser + deps = data.get("dependencies", {}) + dev_deps = data.get("devDependencies", {}) + dependencies["javascript"].extend(list(deps.keys()) + list(dev_deps.keys())) + except Exception: + # Fallback to simple regex if JSON is messy or encoding fails + try: + content = package_json.read_text(encoding='utf-8') + deps = re.findall(r'"([^"]+)":\s*"[^"]*"', content) + dependencies["javascript"].extend(deps) + except Exception: + pass + + return _json(dependencies) + + + + +# ── MAIN / TRANSPORT ────────────────────────────────────────────────── + +def main() -> None: + transport = os.environ.get("MCP_TRANSPORT", "stdio") + mcp.run(transport=transport) + + +if __name__ == "__main__": + main() diff --git a/mcp-servers/project_health/pyproject.toml b/mcp-servers/project_health/pyproject.toml new file mode 100644 index 0000000..5d0c2d9 --- /dev/null +++ b/mcp-servers/project_health/pyproject.toml @@ -0,0 +1,21 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "mcp-server-project_health" +version = "0.1.0" +description = "Analyzes codebase health, including statistics, markers (TODO/FIXME), secrets detection, duplicate files, and dependency analysis." +requires-python = ">=3.11" +dependencies = [ + "mcp>=1.27", + "pydantic>=2.0", + +] + +[project.scripts] +mcp-server-project_health = "app.mcp_server:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["app*"] diff --git a/mcp-servers/time_toolkit/README.md b/mcp-servers/time_toolkit/README.md new file mode 100644 index 0000000..0b751f1 --- /dev/null +++ b/mcp-servers/time_toolkit/README.md @@ -0,0 +1,24 @@ +# MCP Server: time_toolkit + +Utilities for formatting, calculating differences, time arithmetic, and parsing natural language dates. + +## 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/time_toolkit.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 diff --git a/mcp-servers/time_toolkit/app/__init__.py b/mcp-servers/time_toolkit/app/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/mcp-servers/time_toolkit/app/__init__.py diff --git a/mcp-servers/time_toolkit/app/mcp_server.py b/mcp-servers/time_toolkit/app/mcp_server.py new file mode 100644 index 0000000..0f130d5 --- /dev/null +++ b/mcp-servers/time_toolkit/app/mcp_server.py @@ -0,0 +1,270 @@ +"""MCP server for time_toolkit — A toolkit for advanced datetime manipulation and natural language parsing.""" + +from __future__ import annotations + +import json +import os +import re +from datetime import datetime, timedelta, timezone +from enum import Enum +from typing import Annotated, Any +from zoneinfo import ZoneInfo + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +INSTRUCTIONS = """ +time_toolkit provides tools for advanced datetime manipulation and natural language parsing. + +Use it when the task involves: +- Formatting ISO strings into various human-readable or standard formats. +- Calculating the duration between two timestamps in different units. +- Adding or subtracting time intervals from a specific datetime. +- Parsing natural language time expressions (e.g., "tomorrow", "in 2 hours") into ISO timestamps. + +Workflow: +1. parse_natural — convert natural language to an ISO string. +2. format_datetime — format the resulting ISO string for display. +3. add_time — perform arithmetic on the datetime. +4. calculate_duration — find the difference between two points in time. + +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() + +mcp = FastMCP("time_toolkit", instructions=INSTRUCTIONS) + + +def _json(data: Any) -> str: + return json.dumps(data, ensure_ascii=False, indent=2) + + +class OutputFormat(str, Enum): + HUMAN = "human" + ISO = "iso" + RFC2822 = "rfc2822" + SHORT = "short" + + +class DurationUnit(str, Enum): + SECONDS = "seconds" + MINUTES = "minutes" + HOURS = "hours" + DAYS = "days" + WEEKS = "weeks" + MONTHS = "months" + YEARS = "years" + + +# ── TOOL DEFINITIONS ── + +@mcp.tool(name="format_datetime") +async def format_datetime( + iso_string: Annotated[str, Field(description="Input ISO 8601 string.")], + output_format: Annotated[OutputFormat, Field(description="The desired output format: 'human', 'iso', 'rfc2822', or 'short'.")], + target_timezone: Annotated[str, Field(description="IANA timezone name (e.g., 'UTC', 'Europe/Moscow'). Default is 'UTC'.")] = "UTC", +) -> str: + """Formats an ISO 8601 string into a specified format and timezone.""" + try: + dt_str = iso_string.replace("Z", "+00:00") + dt = datetime.fromisoformat(dt_str) + + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + + tz = ZoneInfo(target_timezone) + dt_localized = dt.astimezone(tz) + + if output_format == OutputFormat.ISO: + formatted = dt_localized.isoformat() + elif output_format == OutputFormat.RFC2822: + formatted = dt_localized.strftime("%a, %d %b %Y %H:%M:%S %z") + elif output_format == OutputFormat.SHORT: + formatted = dt_localized.strftime("%d.%m.%Y %H:%M") + else: # human + formatted = dt_localized.strftime("%d %B %Y, %H:%M") + + return _json({ + "formatted": formatted, + "timezone": target_timezone, + "original_iso": iso_string + }) + except Exception as e: + return _json({"error": str(e)}) + + +@mcp.tool(name="calculate_duration") +async def calculate_duration( + start_iso: Annotated[str, Field(description="Start ISO 8601 string.")], + end_iso: Annotated[str, Field(description="End ISO 8601 string.")], + unit: Annotated[DurationUnit, Field(description="The unit for the duration value.")] +) -> str: + """Calculates the duration between two ISO 8601 timestamps in the specified unit.""" + try: + start_dt = datetime.fromisoformat(start_iso.replace("Z", "+00:00")) + end_dt = datetime.fromisoformat(end_iso.replace("Z", "+00:00")) + + diff = end_dt - start_dt + seconds = diff.total_seconds() + + if unit == DurationUnit.SECONDS: + value = seconds + elif unit == DurationUnit.MINUTES: + value = seconds / 60 + elif unit == DurationUnit.HOURS: + value = seconds / 3600 + elif unit == DurationUnit.DAYS: + value = seconds / 86400 + elif unit == DurationUnit.WEEKS: + value = seconds / (86400 * 7) + elif unit == DurationUnit.MONTHS: + value = seconds / (86400 * 30) + elif unit == DurationUnit.YEARS: + value = seconds / (86400 * 365) + else: + raise ValueError(f"Unsupported unit: {unit}") + + return _json({ + "value": round(float(value), 4), + "unit": unit.value, + "start_iso": start_iso, + "end_iso": end_iso + }) + except Exception as e: + return _json({"error": str(e)}) + + +@mcp.tool(name="add_time") +async def add_time( + iso_string: Annotated[str, Field(description="Base ISO 8601 string.")], + value: Annotated[int, Field(description="The amount of time to add (can be negative).")], + unit: Annotated[DurationUnit, Field(description="The unit of the value to add.")] +) -> str: + """Adds or subtracts a specified amount of time from an ISO 8601 timestamp.""" + try: + dt = datetime.fromisoformat(iso_string.replace("Z", "+00:00")) + + if unit == DurationUnit.SECONDS: + delta = timedelta(seconds=value) + elif unit == Tuple[DurationUnit, str] if False else DurationUnit.MINUTES: + delta = timedelta(minutes=value) + elif unit == DurationUnit.MINUTES: + delta = timedelta(minutes=value) + elif unit == DurationUnit.HOURS: + delta = timedelta(hours=value) + elif unit == DurationUnit.DAYS: + delta = timedelta(days=value) + elif unit == DurationUnit.WEEKS: + delta = timedelta(weeks=value) + elif unit == DurationUnit.MONTHS: + delta = timedelta(days=value * 30) + elif unit == DurationUnit.YEARS: + delta = timedelta(days=value * 365) + else: + raise ValueError(f"Unsupported unit: {unit}") + + result_dt = dt + delta + op_str = f"{'+' if value >= 0 else ''}{value} {unit.value}" + + return _json({ + "result_iso": result_dt.isoformat(), + "original_iso": iso_string, + "operation": op_str + }) + except Exception as e: + return _json({"error": str(e)}) + + +@mcp.tool(name="parse_natural") +async def parse_natural( + text: Annotated[str, Field(description="Natural language time expression (e.g., 'tomorrow', 'in 2 hours').")], +) -> str: + """Parses natural language time expressions into ISO 8601 timestamps.""" + try: + now = datetime.now(timezone.utc) + text = text.lower().strip() + + result_dt = None + is_relative = False + + # Absolute simple cases + if text == "now": + result_dt = now + elif text == "today": + result_dt = now.replace(hour=0, minute=0, second=0, microsecond=0) + elif text == "tomorrow": + result_dt = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + elif text == "yesterday": + result_dt = (now - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + + # Relative "in X units" + elif re.match(r"^in (\d+) (second|minute|hour|day|week|month|year)s?$", text): + match = re.match(r"^in (\d+) (second|minute|hour|day|week|month|year)s?$", text) + val = int(match.group(1)) + unit = match.group(2) + is_relative = True + if unit == "second": delta = timedelta(seconds=val) + elif unit == "minute": delta = timedelta(minutes=val) + elif unit == "hour": delta = timedelta(hours=val) + elif unit == "day": delta = timedelta(days=val) + elif unit == "week": delta = timedelta(weeks=val) + elif unit == "month": delta = timedelta(days=val * 30) + elif unit == "year": delta = timedelta(days=val * 365) + result_dt = now + delta + + # Relative "X units ago" + elif reint := re.match(r"^(\d+) (second|minute|hour|day|week|month|year)s? ago$", text): + match = reint + val = int(match.group(1)) + unit = match.group(2) + is_relative = True + if unit == "second": delta = timedelta(seconds=val) + elif unit == "minute": delta = timedelta(minutes=val) + elif unit == "hour": delta = timedelta(hours=val) + elif unit == "day": delta = timedelta(days=val) + elif unit == "week": delta = timedelta(weeks=val) + elif unit == "month": delta = timedelta(days=val * 30) + elif unit == "year": delta = timedelta(days=val * 365) + result_dt = now - delta + + # "next Monday", "last Friday" + elif "next" in text or "last" in text: + days_map = {"monday": 0, "tuesday": 1, "wednesday": 2, "thursday": 3, "friday": 4, "saturday": 5, "sunday": 6} + parts = text.split() + day_name = parts[-1] + if day_name in days_map: + is_relative = True + target_day_idx = days_map[day_name] + current_day_idx = now.weekday() + + if "next" in text: + diff = (target_day_idx - current_day_idx) % 7 + if diff == 0: diff = 7 + result_dt = (now + timedelta(days=diff)).replace(hour=0, minute=0, second=0, microsecond=0) + elif "last" in text: + diff = (current_day_idx - target_day_idx) % 7 + if diff == 0: diff = 7 + result_dt = (now - timedelta(days=diff)).replace(hour=0, minute=0, second=0, microsecond=0) + + if result_dt: + return _json({ + "iso": resultint_dt.isoformat() if (resultint_dt := result_dt) else "", + "parsed_from": text, + "type": "relative" if is_relative else "absolute" + }) + else: + return _json({"error": "could not parse"}) + + except Exception as e: + return _json({"error": str(e)}) + + +# ── MAIN / TRANSPORT ── + +def main() -> None: + transport = os.environ.get("MCP_TRANSPORT", "stdio") + mcp.run(transport=transport) + + +if __name__ == "__main__": + main diff --git a/mcp-servers/time_toolkit/pyproject.toml b/mcp-servers/time_toolkit/pyproject.toml new file mode 100644 index 0000000..859a890 --- /dev/null +++ b/mcp-servers/time_toolkit/pyproject.toml @@ -0,0 +1,21 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "mcp-server-time_toolkit" +version = "0.1.0" +description = "Utilities for formatting, calculating differences, time arithmetic, and parsing natural language dates." +requires-python = ">=3.11" +dependencies = [ + "mcp>=1.27", + "pydantic>=2.0", + +] + +[project.scripts] +mcp-server-time_toolkit = "app.mcp_server:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["app*"]