4fd171b0a4
- Add savearth-workspace MCP server deployment to conscience state. - Refactor .agent/ references to replica-omnisciente/ in operational scripts (setup.sh, savearth-mcp/server.py) and docs (READMEs, DIRECTORY_GUIDE.md, upgrade-conscience workflow). - Document remaining legacy misspellings and stale .agent/ doc refs for follow-up cleanup. - Record conscience upgrade report in data/conscience/ and .aurelio/memory/conscience_upgrade_report.md. [skip ci]
1890 lines
63 KiB
Python
1890 lines
63 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
savearth Unified MCP Server
|
||
|
||
Consolidates device management, telemetry analysis, and infrastructure
|
||
monitoring into a single MCP server. Supports stdio (local) and SSE
|
||
(remote) transports.
|
||
|
||
Usage:
|
||
# Local (stdio — default)
|
||
python3 server.py
|
||
|
||
# Remote (SSE over HTTP)
|
||
python3 server.py --transport sse --port 8080
|
||
|
||
# With auth for SSE
|
||
MCP_AUTH_TOKEN=secret python3 server.py --transport sse
|
||
"""
|
||
|
||
import asyncio
|
||
import glob as globmod
|
||
import json
|
||
import os
|
||
import re
|
||
import shlex
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
from datetime import datetime, timezone, timedelta
|
||
from decimal import Decimal
|
||
from enum import Enum
|
||
from pathlib import Path
|
||
from typing import Optional, List, Dict, Any
|
||
|
||
from pydantic import BaseModel, Field, field_validator, ConfigDict
|
||
from mcp.server.fastmcp import FastMCP, Context
|
||
from mcp.server.fastmcp.server import TransportSecuritySettings
|
||
|
||
# ─── Configuration ───────────────────────────────────────────────────
|
||
|
||
_THIS_DIR = Path(__file__).resolve().parent
|
||
_DEFAULT_PROJECT_DIR = str(_THIS_DIR.parent.parent)
|
||
|
||
PROJECT_DIR = os.environ.get("SAVEARTH_PROJECT_DIR", _DEFAULT_PROJECT_DIR)
|
||
LOGS_DIR = os.path.join(PROJECT_DIR, "logs")
|
||
ANALYZER_SCRIPT = os.path.join(
|
||
PROJECT_DIR, "replica-omnisciente/realms/smart-device-firmware/.aurelio/skills/firmware-log-analyzer/scripts/analyze_log.sh"
|
||
)
|
||
|
||
# InfluxDB — reads from env, falling back to known defaults
|
||
INFLUXDB_URL = os.environ.get("INFLUXDB_URL", "http://172.31.21.178:8086")
|
||
INFLUXDB_TOKEN = os.environ.get("INFLUXDB_TOKEN", "fAclpE9qRE9r9XfRkFf9")
|
||
INFLUXDB_ORG = os.environ.get("INFLUXDB_ORG", "Savearth")
|
||
INFLUXDB_BUCKET = os.environ.get("INFLUXDB_BUCKET", "savearth-iot")
|
||
|
||
# AWS
|
||
AWS_REGION = os.environ.get("AWS_REGION", "eu-north-1")
|
||
DYNAMODB_TABLE = os.environ.get("DYNAMODB_TABLE", "DeviceLogs-v3")
|
||
|
||
# Known fleet
|
||
DEFAULT_DEVICES = [
|
||
"dc:b4:d9:01:58:38", "dc:b4:d9:00:f6:90", "dc:b4:d9:00:f6:d4",
|
||
"10:b4:1d:e1:b5:24", "dc:b4:d9:01:57:a4", "dc:b4:d9:01:57:e0",
|
||
"dc:b4:d9:00:f6:a8", "dc:b4:d9:00:f6:f4", "dc:b4:d9:00:f6:e4",
|
||
"dc:b4:d9:00:f6:44", "dc:b4:d9:01:58:3c", "dc:b4:d9:00:f6:5c",
|
||
"dc:b4:d9:00:f6:c4", "dc:b4:d9:00:f6:94", "dc:b4:d9:01:58:48",
|
||
"dc:b4:d9:00:f6:ec", "dc:b4:d9:01:58:34", "dc:b4:d9:01:58:28",
|
||
"dc:b4:d9:01:57:c4", "10:b4:1d:e1:b5:1c", "10:b4:1d:e1:ac:84",
|
||
"dc:b4:d9:00:f6:a0", "dc:b4:d9:01:58:20", "dc:b4:d9:00:f6:54",
|
||
"10:b4:1d:e1:b4:e4", "dc:b4:d9:00:f6:a4", "10:b4:1d:e1:b5:08",
|
||
"dc:b4:d9:00:f6:28", "10:b4:1d:e1:ac:f4", "dc:b4:d9:00:f6:cc",
|
||
"10:b4:1d:e1:b5:50", "dc:b4:d9:01:57:64", "dc:b4:d9:00:f6:4c",
|
||
"10:b4:1d:e1:ad:30", "dc:b4:d9:01:57:cc", "10:b4:1d:e1:ac:8c",
|
||
"10:b4:1d:e1:ac:a8", "10:b4:1d:e1:b5:64", "dc:b4:d9:01:58:40",
|
||
"dc:b4:d9:01:57:54",
|
||
]
|
||
|
||
# ─── Initialize MCP Server ──────────────────────────────────────────
|
||
|
||
mcp = FastMCP(
|
||
"savearth-mcp",
|
||
# Disable DNS rebinding protection — server runs on a private LAN
|
||
transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
|
||
)
|
||
|
||
|
||
# ─── Enums & Input Models ───────────────────────────────────────────
|
||
|
||
class ResponseFormat(str, Enum):
|
||
MARKDOWN = "markdown"
|
||
JSON = "json"
|
||
|
||
|
||
# ─── Shared Utilities ────────────────────────────────────────────────
|
||
|
||
async def _run_command(
|
||
cmd: List[str],
|
||
cwd: Optional[str] = None,
|
||
timeout: int = 300,
|
||
) -> Dict[str, Any]:
|
||
"""Run a shell command and return structured result."""
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
cwd=cwd or PROJECT_DIR,
|
||
)
|
||
stdout, stderr = await asyncio.wait_for(
|
||
proc.communicate(), timeout=timeout
|
||
)
|
||
return {
|
||
"returncode": proc.returncode,
|
||
"stdout": stdout.decode("utf-8", errors="replace"),
|
||
"stderr": stderr.decode("utf-8", errors="replace"),
|
||
"success": proc.returncode == 0,
|
||
}
|
||
except asyncio.TimeoutError:
|
||
proc.kill()
|
||
return {
|
||
"returncode": -1,
|
||
"stdout": "",
|
||
"stderr": f"Command timed out after {timeout}s",
|
||
"success": False,
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"returncode": -1,
|
||
"stdout": "",
|
||
"stderr": str(e),
|
||
"success": False,
|
||
}
|
||
|
||
|
||
def _human_size(nbytes: int) -> str:
|
||
for unit in ["B", "KB", "MB", "GB"]:
|
||
if abs(nbytes) < 1024:
|
||
return f"{nbytes:.1f} {unit}"
|
||
nbytes /= 1024.0
|
||
return f"{nbytes:.1f} TB"
|
||
|
||
|
||
def _extract_warnings(stderr: str) -> str:
|
||
warnings = [l for l in stderr.split("\n") if "warning:" in l.lower()]
|
||
if not warnings:
|
||
return "No warnings"
|
||
return "\n".join(warnings[:20])
|
||
|
||
|
||
def _flux_query(query: str, timeout: int = 15) -> str:
|
||
"""Execute a Flux query against InfluxDB. Returns CSV text."""
|
||
url = f"{INFLUXDB_URL}/api/v2/query?org={INFLUXDB_ORG}"
|
||
headers = {
|
||
"Authorization": f"Token {INFLUXDB_TOKEN}",
|
||
"Content-Type": "application/vnd.flux",
|
||
"Accept": "application/csv",
|
||
}
|
||
req = urllib.request.Request(url, data=query.encode(), headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
return resp.read().decode()
|
||
except urllib.error.URLError as e:
|
||
raise ConnectionError(
|
||
f"InfluxDB unreachable at {INFLUXDB_URL}: {e}\n"
|
||
"If running locally, start an SSM tunnel first."
|
||
)
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# DEVICE TOOLS — Build, Flash, Monitor, Analyze
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
@mcp.tool(
|
||
name="device_build_firmware",
|
||
annotations={
|
||
"title": "Build ESP32-S3 Firmware",
|
||
"readOnlyHint": False,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_build_firmware(
|
||
target: Optional[str] = None,
|
||
verbose: bool = False,
|
||
sdkconfig_defaults: Optional[str] = None,
|
||
ctx: Context = None,
|
||
) -> str:
|
||
"""Build the ESP32-S3 firmware using idf.py.
|
||
|
||
Args:
|
||
target: Build target — 'app' (default), 'test_<component>' for unit tests, or 'clean'.
|
||
verbose: Enable verbose build output.
|
||
sdkconfig_defaults: Optional sdkconfig defaults files. Multiple files can be specified
|
||
separated by semicolons (e.g., 'sdkconfig.defaults;sdkconfig.ci.release').
|
||
When provided, causes a reconfigure and full rebuild.
|
||
"""
|
||
if ctx:
|
||
await ctx.report_progress(0.1, "Starting build...")
|
||
|
||
cmd = ["idf.py"]
|
||
if target == "clean":
|
||
cmd.append("fullclean")
|
||
elif target and target.startswith("test_"):
|
||
cmd.extend(["-T", target, "build"])
|
||
else:
|
||
cmd.append("build")
|
||
|
||
if sdkconfig_defaults and sdkconfig_defaults.strip():
|
||
cmd.extend(["-D", f"SDKCONFIG_DEFAULTS={sdkconfig_defaults}"])
|
||
|
||
if verbose:
|
||
cmd.append("-v")
|
||
|
||
result = await _run_command(cmd, cwd=PROJECT_DIR, timeout=600)
|
||
|
||
if ctx:
|
||
await ctx.report_progress(1.0, "Build complete")
|
||
|
||
if result["success"]:
|
||
output = result["stdout"]
|
||
size_line = ""
|
||
for line in output.split("\n"):
|
||
if "Binary size" in line or "Total sizes" in line:
|
||
size_line = line.strip()
|
||
break
|
||
|
||
return (
|
||
f"## ✅ Build Successful\n\n"
|
||
f"```\n{size_line or 'Build completed.'}\n```\n\n"
|
||
f"**Warnings:**\n```\n"
|
||
f"{_extract_warnings(result['stderr'])}\n```"
|
||
)
|
||
else:
|
||
errors = result["stderr"][-2000:]
|
||
return (
|
||
f"## ❌ Build Failed\n\n"
|
||
f"**Exit code:** {result['returncode']}\n\n"
|
||
f"**Errors:**\n```\n{errors}\n```"
|
||
)
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_flash",
|
||
annotations={
|
||
"title": "Flash Firmware to Device",
|
||
"readOnlyHint": False,
|
||
"destructiveHint": True,
|
||
"idempotentHint": False,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_flash(
|
||
port: str = "/dev/ttyUSB0",
|
||
baud: int = 460800,
|
||
ctx: Context = None,
|
||
) -> str:
|
||
"""Flash compiled firmware to a connected ESP32-S3 device.
|
||
|
||
WARNING: This overwrites the device's current firmware.
|
||
|
||
Args:
|
||
port: Serial port (e.g., /dev/ttyUSB0, /dev/ttyACM0).
|
||
baud: Baud rate for flashing (9600–2000000).
|
||
"""
|
||
if ctx:
|
||
await ctx.report_progress(0.1, f"Flashing to {port}...")
|
||
|
||
cmd = ["idf.py", "-p", port, "-b", str(baud), "flash"]
|
||
result = await _run_command(cmd, cwd=PROJECT_DIR, timeout=120)
|
||
|
||
if result["success"]:
|
||
return (
|
||
f"## ✅ Flash Successful\n\n"
|
||
f"- **Port:** {port}\n"
|
||
f"- **Baud:** {baud}\n\n"
|
||
f"Device ready. Use `device_monitor` to view serial output."
|
||
)
|
||
else:
|
||
return (
|
||
f"## ❌ Flash Failed\n\n"
|
||
f"**Port:** {port}\n\n"
|
||
f"**Error:**\n```\n{result['stderr'][-1000:]}\n```\n\n"
|
||
f"**Troubleshooting:**\n"
|
||
f"- Check USB cable is connected\n"
|
||
f"- Try holding BOOT button while pressing RESET\n"
|
||
f"- Check permissions: `sudo chmod 666 {port}`"
|
||
)
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_monitor",
|
||
annotations={
|
||
"title": "Capture Serial Monitor Output",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": False,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_monitor(
|
||
port: str = "/dev/ttyUSB0",
|
||
duration_seconds: int = 30,
|
||
save_to_file: bool = True,
|
||
ctx: Context = None,
|
||
) -> str:
|
||
"""Capture serial output from a connected ESP32-S3 device for N seconds.
|
||
|
||
Args:
|
||
port: Serial port for the device.
|
||
duration_seconds: Capture duration (5–600 seconds).
|
||
save_to_file: Save captured output to logs/ directory.
|
||
"""
|
||
duration_seconds = max(5, min(600, duration_seconds))
|
||
|
||
if ctx:
|
||
await ctx.report_progress(0.1, f"Capturing {duration_seconds}s from {port}...")
|
||
|
||
cmd = [
|
||
"timeout", str(duration_seconds),
|
||
"idf.py", "-p", port, "monitor", "--no-reset",
|
||
]
|
||
|
||
result = await _run_command(cmd, cwd=PROJECT_DIR, timeout=duration_seconds + 10)
|
||
output = result["stdout"]
|
||
lines = output.count("\n")
|
||
|
||
if save_to_file:
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
filename = f"capture_{timestamp}.log"
|
||
filepath = os.path.join(LOGS_DIR, filename)
|
||
os.makedirs(LOGS_DIR, exist_ok=True)
|
||
with open(filepath, "w") as f:
|
||
f.write(output)
|
||
|
||
return (
|
||
f"## 📡 Serial Capture Complete\n\n"
|
||
f"- **Duration:** {duration_seconds}s\n"
|
||
f"- **Lines captured:** {lines}\n"
|
||
f"- **Saved to:** `{filepath}`\n\n"
|
||
f"Use `device_analyze_log` with this file to check for bugs.\n\n"
|
||
f"**Last 20 lines:**\n```\n"
|
||
f"{chr(10).join(output.split(chr(10))[-20:])}\n```"
|
||
)
|
||
else:
|
||
return (
|
||
f"## 📡 Serial Capture ({duration_seconds}s, {lines} lines)\n\n"
|
||
f"```\n{output[-3000:]}\n```"
|
||
)
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_list_logs",
|
||
annotations={
|
||
"title": "List Log Files",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_list_logs(
|
||
pattern: str = "*",
|
||
response_format: str = "markdown",
|
||
) -> str:
|
||
"""List available log and analysis files in the logs/ directory.
|
||
|
||
Args:
|
||
pattern: Glob pattern to filter (e.g., '*.log', 'Firmware-*').
|
||
response_format: Output format — 'markdown' or 'json'.
|
||
"""
|
||
|
||
pattern_path = os.path.join(LOGS_DIR, pattern)
|
||
files = sorted(globmod.glob(pattern_path))
|
||
if not files:
|
||
return f"No files matching `{pattern}` in `{LOGS_DIR}`"
|
||
|
||
file_infos = []
|
||
for f in files:
|
||
if os.path.isfile(f):
|
||
stat = os.stat(f)
|
||
file_infos.append({
|
||
"name": os.path.basename(f),
|
||
"path": f,
|
||
"size_bytes": stat.st_size,
|
||
"size_human": _human_size(stat.st_size),
|
||
})
|
||
|
||
if response_format == "json":
|
||
return json.dumps(file_infos, indent=2, default=str)
|
||
|
||
lines = [f"## 📂 Log Files ({len(file_infos)} files)\n"]
|
||
for fi in file_infos:
|
||
lines.append(f"- **{fi['name']}** — {fi['size_human']}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_analyze_log",
|
||
annotations={
|
||
"title": "Analyze Firmware Log for Known Bugs",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_analyze_log(
|
||
log_file: str,
|
||
ctx: Context = None,
|
||
) -> str:
|
||
"""Run the firmware-log-analyzer on a log file. Checks for all known bug patterns.
|
||
|
||
Args:
|
||
log_file: Path to the log file (relative to logs/ or absolute).
|
||
"""
|
||
if not os.path.isabs(log_file):
|
||
log_file = os.path.join(LOGS_DIR, log_file)
|
||
|
||
if not os.path.isfile(log_file):
|
||
return f"Error: File not found: `{log_file}`"
|
||
|
||
if ctx:
|
||
await ctx.report_progress(0.2, f"Analyzing {os.path.basename(log_file)}...")
|
||
|
||
if os.path.isfile(ANALYZER_SCRIPT):
|
||
result = await _run_command(
|
||
["bash", ANALYZER_SCRIPT, log_file], timeout=60
|
||
)
|
||
output = result["stdout"] if result["stdout"] else result["stderr"]
|
||
else:
|
||
output = await _inline_analysis(log_file)
|
||
|
||
return f"## 🔍 Log Analysis: `{os.path.basename(log_file)}`\n\n```\n{output}\n```"
|
||
|
||
|
||
async def _inline_analysis(log_file: str) -> str:
|
||
"""Fallback inline analysis if analyze_log.sh is not available."""
|
||
patterns = {
|
||
"BUG-01 Config mode stuck": r"Config Mode Run: enabled=1|Config mode is ENABLED",
|
||
"BUG-03 SoftAP active": r"wifi:mode : sta \+ softAP",
|
||
"BUG-08 data_storage race": r"Failed to unlink|Has open FD",
|
||
"BUG-10 TLS force-kill": r"Force-killing stuck TLS",
|
||
"BUG-15 Beacon timeout": r"bcn_timeout",
|
||
"BUG-09 MQTT disconnect": r"MQTTRecvFailed",
|
||
"BUG-06 DNS before DHCP": r"getaddrinfo.*returns 202",
|
||
}
|
||
|
||
results = []
|
||
for name, pattern in patterns.items():
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"grep", "-ciE", pattern, log_file,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
stdout, _ = await proc.communicate()
|
||
count = int(stdout.decode().strip() or "0")
|
||
if count > 0:
|
||
results.append(f" {name}: {count} occurrences")
|
||
|
||
if results:
|
||
return f"FOUND {len(results)} bug pattern(s):\n" + "\n".join(results)
|
||
return "HEALTHY: No known bugs detected."
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_check_ota",
|
||
annotations={
|
||
"title": "Check OTA Firmware Versions on S3",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def device_check_ota() -> str:
|
||
"""Check the S3 firmware bucket for available OTA versions."""
|
||
script = os.path.join(PROJECT_DIR, "scripts/check_firmware.py")
|
||
|
||
if os.path.isfile(script):
|
||
result = await _run_command(
|
||
["python3", script], cwd=PROJECT_DIR, timeout=30
|
||
)
|
||
if result["success"]:
|
||
return f"## 🔄 OTA Firmware Status\n\n```\n{result['stdout']}\n```"
|
||
else:
|
||
return (
|
||
f"## ❌ OTA Check Failed\n\n```\n{result['stderr']}\n```\n\n"
|
||
"Ensure AWS credentials are configured."
|
||
)
|
||
else:
|
||
return "Error: `scripts/check_firmware.py` not found."
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_sdkconfig_get",
|
||
annotations={
|
||
"title": "Read sdkconfig Values",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_sdkconfig_get(
|
||
key: Optional[str] = None,
|
||
response_format: str = "markdown",
|
||
) -> str:
|
||
"""Read configuration values from sdkconfig.
|
||
|
||
Args:
|
||
key: Config key to search (e.g., 'CONFIG_FREERTOS'). If empty, returns summary.
|
||
response_format: Output format — 'markdown' or 'json'.
|
||
"""
|
||
sdkconfig_path = os.path.join(PROJECT_DIR, "sdkconfig")
|
||
if not os.path.isfile(sdkconfig_path):
|
||
return "Error: `sdkconfig` not found. Run `device_build_firmware` first."
|
||
|
||
with open(sdkconfig_path, "r") as f:
|
||
lines = f.readlines()
|
||
|
||
if key:
|
||
matched = [
|
||
l.strip()
|
||
for l in lines
|
||
if key.upper() in l.upper() and not l.startswith("#")
|
||
]
|
||
if not matched:
|
||
return f"No config matching `{key}` found."
|
||
if response_format == "json":
|
||
configs = {}
|
||
for m in matched:
|
||
if "=" in m:
|
||
k, v = m.split("=", 1)
|
||
configs[k] = v
|
||
return json.dumps(configs, indent=2)
|
||
return f"## sdkconfig: `{key}`\n\n```\n" + "\n".join(matched) + "\n```"
|
||
else:
|
||
non_comment = [l.strip() for l in lines if l.strip() and not l.startswith("#")]
|
||
return (
|
||
f"## sdkconfig ({len(non_comment)} entries)\n\n"
|
||
"Use `key` parameter to filter. Example: `CONFIG_MQTT`"
|
||
)
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_sdkconfig_set",
|
||
annotations={
|
||
"title": "Set sdkconfig Value",
|
||
"readOnlyHint": False,
|
||
"destructiveHint": True,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_sdkconfig_set(
|
||
key: str,
|
||
value: str,
|
||
ctx: Context = None,
|
||
) -> str:
|
||
"""Set a configuration value in sdkconfig.defaults.
|
||
|
||
Modifies the defaults file (not sdkconfig directly) so the change
|
||
persists across clean builds. Requires a rebuild to take effect.
|
||
|
||
WARNING: Changing build config can affect firmware behavior.
|
||
|
||
Args:
|
||
key: Config key (e.g., 'CONFIG_MQTT_KEEPALIVE').
|
||
value: Config value to set.
|
||
"""
|
||
defaults_path = os.path.join(PROJECT_DIR, "sdkconfig.defaults")
|
||
if not os.path.isfile(defaults_path):
|
||
return "Error: `sdkconfig.defaults` not found."
|
||
|
||
with open(defaults_path, "r") as f:
|
||
content = f.read()
|
||
|
||
new_line = f"{key}={value}"
|
||
|
||
if key in content:
|
||
lines = content.split("\n")
|
||
updated = []
|
||
for line in lines:
|
||
if line.startswith(key + "="):
|
||
if ctx:
|
||
await ctx.log_info(f"Replacing: {line} → {new_line}")
|
||
updated.append(new_line)
|
||
else:
|
||
updated.append(line)
|
||
content = "\n".join(updated)
|
||
else:
|
||
content += f"\n{new_line}\n"
|
||
if ctx:
|
||
await ctx.log_info(f"Appending: {new_line}")
|
||
|
||
with open(defaults_path, "w") as f:
|
||
f.write(content)
|
||
|
||
return (
|
||
f"## ✅ sdkconfig Updated\n\n"
|
||
f"- **Key:** `{key}`\n"
|
||
f"- **Value:** `{value}`\n"
|
||
f"- **File:** `sdkconfig.defaults`\n\n"
|
||
f"> Run `device_build_firmware` to apply changes."
|
||
)
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_list_serial_ports",
|
||
annotations={
|
||
"title": "List ESP Serial Ports",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_list_serial_ports() -> str:
|
||
"""List available serial ports for ESP devices.
|
||
|
||
Uses pyserial for auto-detection with fallback to common device paths.
|
||
"""
|
||
# Try pyserial first
|
||
result = await _run_command(
|
||
["python3", "-m", "serial.tools.list_ports", "-v"], timeout=10
|
||
)
|
||
|
||
if result["success"] and result["stdout"].strip():
|
||
ports = result["stdout"].strip()
|
||
count = len([l for l in ports.split("\n") if l.strip() and not l.startswith(" ")])
|
||
return (
|
||
f"## 🔌 Serial Ports ({count} found)\n\n"
|
||
f"```\n{ports}\n```"
|
||
)
|
||
|
||
# Fallback: glob common ESP device paths
|
||
patterns = ["/dev/ttyACM*", "/dev/ttyUSB*"]
|
||
found = []
|
||
for pat in patterns:
|
||
found.extend(sorted(globmod.glob(pat)))
|
||
|
||
if found:
|
||
lines = [f"## 🔌 Serial Ports ({len(found)} found)\n"]
|
||
for port in found:
|
||
lines.append(f"- `{port}`")
|
||
lines.append("\n> *Detected via glob — install `pyserial` for detailed info.*")
|
||
return "\n".join(lines)
|
||
|
||
return (
|
||
"## 🔌 No Serial Ports Found\n\n"
|
||
"No ESP devices detected. Check:\n"
|
||
"- USB cable is connected\n"
|
||
"- Device is powered on\n"
|
||
"- User has permission (`sudo chmod 666 /dev/ttyACM0`)"
|
||
)
|
||
|
||
|
||
# Known ESP32 target chips
|
||
_VALID_TARGETS = [
|
||
"esp32", "esp32s2", "esp32s3", "esp32c2", "esp32c3",
|
||
"esp32c5", "esp32c6", "esp32h2", "esp32p4",
|
||
]
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_set_target",
|
||
annotations={
|
||
"title": "Set ESP Target Chip",
|
||
"readOnlyHint": False,
|
||
"destructiveHint": True,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_set_target(
|
||
target: str,
|
||
ctx: Context = None,
|
||
) -> str:
|
||
"""Set the target chip for the ESP-IDF project. Equivalent to `idf.py set-target`.
|
||
|
||
WARNING: This triggers a full clean rebuild.
|
||
|
||
Args:
|
||
target: Lowercase target name (e.g., 'esp32', 'esp32s3', 'esp32c3', 'esp32c6').
|
||
"""
|
||
target = target.lower().strip()
|
||
if target not in _VALID_TARGETS:
|
||
return (
|
||
f"## ❌ Invalid Target\n\n"
|
||
f"`{target}` is not a recognized ESP target.\n\n"
|
||
f"**Valid targets:** {', '.join(f'`{t}`' for t in _VALID_TARGETS)}"
|
||
)
|
||
|
||
if ctx:
|
||
await ctx.report_progress(0.1, f"Setting target to {target}...")
|
||
|
||
cmd = ["idf.py", "set-target", target]
|
||
result = await _run_command(cmd, cwd=PROJECT_DIR, timeout=120)
|
||
|
||
if result["success"]:
|
||
return (
|
||
f"## ✅ Target Set: `{target}`\n\n"
|
||
f"The project is now configured for **{target.upper()}**.\n\n"
|
||
f"> Run `device_build_firmware` to build for this target."
|
||
)
|
||
else:
|
||
return (
|
||
f"## ❌ Set Target Failed\n\n"
|
||
f"**Target:** {target}\n"
|
||
f"**Exit code:** {result['returncode']}\n\n"
|
||
f"**Error:**\n```\n{result['stderr'][-1500:]}\n```"
|
||
)
|
||
|
||
|
||
@mcp.tool(
|
||
name="device_run_pytest",
|
||
annotations={
|
||
"title": "Run Pytest Tests",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": False,
|
||
},
|
||
)
|
||
async def device_run_pytest(
|
||
test_path: str = ".",
|
||
pytest_args: str = "",
|
||
ctx: Context = None,
|
||
) -> str:
|
||
"""Run pytest tests for the project. Supports pytest-embedded for ESP-IDF testing.
|
||
|
||
Uses pytest-embedded (https://docs.espressif.com/projects/pytest-embedded),
|
||
a pytest plugin for embedded testing on ESP32 targets.
|
||
|
||
Args:
|
||
test_path: Path to test file or directory, relative to project root
|
||
(default: '.', runs all tests). Examples: 'test/host', 'test/apps'.
|
||
pytest_args: Additional pytest arguments. Common options:
|
||
-v: Verbose output
|
||
-k EXPRESSION: Run tests matching expression
|
||
-m MARKER: Run tests with specific marker
|
||
--target TARGET: ESP target (esp32, esp32s3, etc.)
|
||
--sdkconfig NAME: sdkconfig config name
|
||
--junitxml=FILE: Generate JUnit XML report
|
||
"""
|
||
start_time = time.time()
|
||
|
||
if ctx:
|
||
await ctx.report_progress(0.1, f"Running pytest on {test_path}...")
|
||
|
||
# Build the command — use shell for argument splitting
|
||
pytest_cmd = f"pytest {shlex.quote(test_path)}"
|
||
if pytest_args:
|
||
pytest_cmd += f" {pytest_args}"
|
||
|
||
cmd = ["bash", "-c", pytest_cmd]
|
||
result = await _run_command(cmd, cwd=PROJECT_DIR, timeout=600)
|
||
|
||
elapsed = time.time() - start_time
|
||
elapsed_min = int(elapsed // 60)
|
||
elapsed_sec = elapsed % 60
|
||
timing = f"{elapsed_min}m {elapsed_sec:.1f}s" if elapsed_min else f"{elapsed_sec:.1f}s"
|
||
|
||
if ctx:
|
||
await ctx.report_progress(1.0, "Tests complete")
|
||
|
||
output = result["stdout"]
|
||
|
||
# Extract summary line (e.g., "===== 8 passed in 0.02s =====")
|
||
summary_line = ""
|
||
for line in reversed(output.split("\n")):
|
||
if "passed" in line or "failed" in line or "error" in line:
|
||
summary_line = line.strip()
|
||
break
|
||
|
||
if result["success"]:
|
||
return (
|
||
f"## ✅ Tests Passed\n\n"
|
||
f"- **Path:** `{test_path}`\n"
|
||
f"- **Duration:** {timing}\n"
|
||
f"- **Summary:** `{summary_line}`\n\n"
|
||
f"**Output (last 3000 chars):**\n```\n{output[-3000:]}\n```"
|
||
)
|
||
else:
|
||
stderr_snippet = result["stderr"][-1000:] if result["stderr"] else ""
|
||
return (
|
||
f"## ❌ Tests Failed\n\n"
|
||
f"- **Path:** `{test_path}`\n"
|
||
f"- **Duration:** {timing}\n"
|
||
f"- **Exit code:** {result['returncode']}\n"
|
||
f"- **Summary:** `{summary_line}`\n\n"
|
||
f"**Output (last 3000 chars):**\n```\n{output[-3000:]}\n```"
|
||
+ (f"\n\n**Stderr:**\n```\n{stderr_snippet}\n```" if stderr_snippet else "")
|
||
)
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# TELEMETRY TOOLS — InfluxDB Queries
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
@mcp.tool(
|
||
name="telemetry_query",
|
||
annotations={
|
||
"title": "Execute Raw Flux Query",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def telemetry_query(query: str) -> str:
|
||
"""Execute a raw Flux query against the InfluxDB database.
|
||
|
||
Args:
|
||
query: Flux query string.
|
||
"""
|
||
try:
|
||
csv_data = _flux_query(query)
|
||
return f"## 📊 Query Results\n\n```csv\n{csv_data}\n```"
|
||
except ConnectionError as e:
|
||
return f"## ❌ Query Failed\n\n**Error:**\n```\n{e}\n```"
|
||
except Exception as e:
|
||
return f"## ❌ Query Failed\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="telemetry_device_status",
|
||
annotations={
|
||
"title": "Get Device Telemetry Status",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def telemetry_device_status(
|
||
device_id: str,
|
||
hours: int = 24,
|
||
) -> str:
|
||
"""Get the latest telemetry status for a specific device.
|
||
|
||
Args:
|
||
device_id: Device ID / Client ID (MAC address).
|
||
hours: Lookback window in hours (default 24).
|
||
"""
|
||
query = f"""
|
||
from(bucket: "{INFLUXDB_BUCKET}")
|
||
|> range(start: -{hours}h)
|
||
|> filter(fn: (r) => r["_measurement"] == "telemetry")
|
||
|> filter(fn: (r) => r["client_id"] == "{device_id}")
|
||
|> last()
|
||
|> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||
"""
|
||
try:
|
||
csv_data = _flux_query(query)
|
||
return f"## 📱 Device Status: {device_id}\n\n```csv\n{csv_data}\n```"
|
||
except ConnectionError as e:
|
||
return f"## ❌ Failed to get status\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="telemetry_volume_history",
|
||
annotations={
|
||
"title": "Get Water Volume History",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def telemetry_volume_history(
|
||
device_id: str,
|
||
hours: int = 24,
|
||
) -> str:
|
||
"""Get water volume consumption history for a device (hourly aggregation).
|
||
|
||
Args:
|
||
device_id: Device ID.
|
||
hours: Lookback hours (default 24).
|
||
"""
|
||
query = f"""
|
||
from(bucket: "{INFLUXDB_BUCKET}")
|
||
|> range(start: -{hours}h)
|
||
|> filter(fn: (r) => r["_measurement"] == "telemetry")
|
||
|> filter(fn: (r) => r["client_id"] == "{device_id}")
|
||
|> filter(fn: (r) => r["_field"] == "total_liters")
|
||
|> aggregateWindow(every: 1h, fn: max, createEmpty: false)
|
||
|> yield(name: "hourly_max")
|
||
"""
|
||
try:
|
||
csv_data = _flux_query(query)
|
||
return f"## 💧 Volume History: {device_id}\n\n```csv\n{csv_data}\n```"
|
||
except ConnectionError as e:
|
||
return f"## ❌ Failed to get volume history\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="telemetry_connection_health",
|
||
annotations={
|
||
"title": "Analyze Connection Reliability",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def telemetry_connection_health(
|
||
device_id: Optional[str] = None,
|
||
hours: int = 24,
|
||
) -> str:
|
||
"""Analyze MQTT connection drops and reliability.
|
||
|
||
Args:
|
||
device_id: Filter by device ID (optional — omit for fleet-wide).
|
||
hours: Lookback hours (default 24).
|
||
"""
|
||
device_filter = ""
|
||
if device_id:
|
||
device_filter = f'\n |> filter(fn: (r) => r["client_id"] == "{device_id}")'
|
||
|
||
query = f"""
|
||
from(bucket: "{INFLUXDB_BUCKET}")
|
||
|> range(start: -{hours}h)
|
||
|> filter(fn: (r) => r["_measurement"] == "mqtt_events"){device_filter}
|
||
|> filter(fn: (r) => r["_field"] == "event_type")
|
||
|> group(columns: ["client_id", "_value"])
|
||
|> count()
|
||
|> group(columns: ["client_id"])
|
||
|> pivot(rowKey:["client_id"], columnKey: ["_value"], valueColumn: "_field")
|
||
"""
|
||
try:
|
||
csv_data = _flux_query(query)
|
||
return f"## 🔌 Connection Health\n\n```csv\n{csv_data}\n```"
|
||
except ConnectionError as e:
|
||
return f"## ❌ Failed to analyze health\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="telemetry_fleet_overview",
|
||
annotations={
|
||
"title": "Fleet Overview — Active Devices",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def telemetry_fleet_overview(hours: int = 24) -> str:
|
||
"""List all devices that reported telemetry recently with their last-seen time.
|
||
|
||
Args:
|
||
hours: Lookback hours (default 24).
|
||
"""
|
||
query = f"""
|
||
from(bucket: "{INFLUXDB_BUCKET}")
|
||
|> range(start: -{hours}h)
|
||
|> filter(fn: (r) => r["_measurement"] == "telemetry")
|
||
|> group(columns: ["client_id"])
|
||
|> last()
|
||
|> keep(columns: ["client_id", "_time"])
|
||
|> sort(columns: ["_time"], desc: true)
|
||
"""
|
||
try:
|
||
csv_data = _flux_query(query)
|
||
return (
|
||
f"## 🌍 Fleet Overview (last {hours}h)\n\n"
|
||
f"```csv\n{csv_data}\n```\n\n"
|
||
f"**Known fleet size:** {len(DEFAULT_DEVICES)} devices"
|
||
)
|
||
except ConnectionError as e:
|
||
return f"## ❌ Failed to get fleet overview\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="telemetry_shower_sessions",
|
||
annotations={
|
||
"title": "Analyze Shower Sessions",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def telemetry_shower_sessions(
|
||
device_id: str,
|
||
days: int = 7,
|
||
) -> str:
|
||
"""Analyze shower sessions — daily volume totals and session detection.
|
||
|
||
Args:
|
||
device_id: Device ID.
|
||
days: Lookback days (default 7).
|
||
"""
|
||
query = f"""
|
||
from(bucket: "{INFLUXDB_BUCKET}")
|
||
|> range(start: -{days}d)
|
||
|> filter(fn: (r) => r["_measurement"] == "telemetry")
|
||
|> filter(fn: (r) => r["client_id"] == "{device_id}")
|
||
|> filter(fn: (r) => r["_field"] == "total_liters")
|
||
|> aggregateWindow(every: 1d, fn: max, createEmpty: false)
|
||
|> difference()
|
||
|> yield(name: "daily_usage")
|
||
"""
|
||
try:
|
||
csv_data = _flux_query(query)
|
||
return (
|
||
f"## 🚿 Shower Sessions: {device_id} (last {days} days)\n\n"
|
||
f"```csv\n{csv_data}\n```"
|
||
)
|
||
except ConnectionError as e:
|
||
return f"## ❌ Failed to analyze sessions\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# INFRASTRUCTURE TOOLS — DynamoDB, Diagnostics, Fleet Health
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
@mcp.tool(
|
||
name="infra_dynamo_logs",
|
||
annotations={
|
||
"title": "Browse Device Logs in DynamoDB",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def infra_dynamo_logs(
|
||
device_id: str,
|
||
minutes: int = 30,
|
||
level_filter: Optional[str] = None,
|
||
keyword: Optional[str] = None,
|
||
) -> str:
|
||
"""Browse device logs from DynamoDB (DeviceLogs-v3).
|
||
|
||
Args:
|
||
device_id: Device ID (MAC address).
|
||
minutes: Lookback window in minutes (default 30).
|
||
level_filter: Filter by log level — 'E' (error), 'W' (warning), 'I' (info).
|
||
keyword: Filter log messages containing this keyword (case-insensitive).
|
||
"""
|
||
try:
|
||
import boto3
|
||
from boto3.dynamodb.conditions import Key
|
||
except ImportError:
|
||
return "Error: `boto3` not installed."
|
||
|
||
try:
|
||
dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION)
|
||
table = dynamodb.Table(DYNAMODB_TABLE)
|
||
now = datetime.now(timezone.utc)
|
||
start_ms = int((now - timedelta(minutes=minutes)).timestamp() * 1000)
|
||
|
||
kce = Key("device_id").eq(device_id) & Key("timestamp").gt(
|
||
Decimal(str(start_ms))
|
||
)
|
||
resp = table.query(KeyConditionExpression=kce, ScanIndexForward=True)
|
||
items = resp.get("Items", [])
|
||
|
||
while "LastEvaluatedKey" in resp:
|
||
resp = table.query(
|
||
KeyConditionExpression=kce,
|
||
ScanIndexForward=True,
|
||
ExclusiveStartKey=resp["LastEvaluatedKey"],
|
||
)
|
||
items.extend(resp.get("Items", []))
|
||
|
||
# Apply filters
|
||
if level_filter:
|
||
items = [i for i in items if i.get("level", "") == level_filter.upper()]
|
||
if keyword:
|
||
kw = keyword.lower()
|
||
items = [
|
||
i for i in items
|
||
if kw in str(i.get("msg", "")).lower()
|
||
or kw in str(i.get("tag", "")).lower()
|
||
]
|
||
|
||
if not items:
|
||
return (
|
||
f"## 📋 No Logs Found\n\n"
|
||
f"**Device:** {device_id}\n"
|
||
f"**Window:** last {minutes}m\n"
|
||
f"**Filters:** level={level_filter or 'any'}, keyword={keyword or 'none'}"
|
||
)
|
||
|
||
# Format output
|
||
error_count = sum(1 for i in items if i.get("level") == "E")
|
||
warn_count = sum(1 for i in items if i.get("level") == "W")
|
||
|
||
# Show last 50 entries
|
||
display = items[-50:]
|
||
lines = []
|
||
for item in display:
|
||
ts = int(item.get("timestamp", 0))
|
||
ts_str = datetime.fromtimestamp(ts / 1000, tz=timezone.utc).strftime(
|
||
"%H:%M:%S"
|
||
)
|
||
level = item.get("level", "?")
|
||
tag = item.get("tag", "")
|
||
msg = str(item.get("msg", ""))[:150]
|
||
lines.append(f"{ts_str} [{level}] {tag}: {msg}")
|
||
|
||
return (
|
||
f"## 📋 Device Logs: {device_id}\n\n"
|
||
f"- **Total:** {len(items)} entries (showing last {len(display)})\n"
|
||
f"- **Errors:** {error_count} | **Warnings:** {warn_count}\n"
|
||
f"- **Window:** last {minutes}m\n\n"
|
||
f"```\n" + "\n".join(lines) + "\n```"
|
||
)
|
||
|
||
except Exception as e:
|
||
return f"## ❌ DynamoDB Query Failed\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="infra_fsm_transitions",
|
||
annotations={
|
||
"title": "Analyze FSM State Transitions",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def infra_fsm_transitions(
|
||
device_id: str,
|
||
minutes: int = 60,
|
||
) -> str:
|
||
"""Parse FSM state transitions from DynamoDB logs. Detects stuck states.
|
||
|
||
Args:
|
||
device_id: Device ID (MAC address).
|
||
minutes: Lookback window in minutes (default 60).
|
||
"""
|
||
try:
|
||
import boto3
|
||
from boto3.dynamodb.conditions import Key
|
||
except ImportError:
|
||
return "Error: `boto3` not installed."
|
||
|
||
try:
|
||
dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION)
|
||
table = dynamodb.Table(DYNAMODB_TABLE)
|
||
now = datetime.now(timezone.utc)
|
||
start_ms = int((now - timedelta(minutes=minutes)).timestamp() * 1000)
|
||
|
||
kce = Key("device_id").eq(device_id) & Key("timestamp").gt(
|
||
Decimal(str(start_ms))
|
||
)
|
||
resp = table.query(KeyConditionExpression=kce, ScanIndexForward=True)
|
||
items = resp.get("Items", [])
|
||
|
||
while "LastEvaluatedKey" in resp:
|
||
resp = table.query(
|
||
KeyConditionExpression=kce,
|
||
ScanIndexForward=True,
|
||
ExclusiveStartKey=resp["LastEvaluatedKey"],
|
||
)
|
||
items.extend(resp.get("Items", []))
|
||
|
||
transition_pattern = re.compile(
|
||
r"STATE TRANSITION.*?(\w+)\s*→\s*(\w+)|"
|
||
r"(\w+)\s*->\s*(\w+)"
|
||
)
|
||
|
||
transitions = []
|
||
for item in items:
|
||
msg = str(item.get("msg", ""))
|
||
ts = int(item.get("timestamp", 0))
|
||
|
||
match = transition_pattern.search(msg)
|
||
if match:
|
||
from_state = match.group(1) or match.group(3)
|
||
to_state = match.group(2) or match.group(4)
|
||
ts_str = datetime.fromtimestamp(
|
||
ts / 1000, tz=timezone.utc
|
||
).strftime("%H:%M:%S")
|
||
transitions.append(f"{ts_str} {from_state} → {to_state}")
|
||
|
||
if not transitions:
|
||
return (
|
||
f"## 🔄 FSM Transitions: {device_id}\n\n"
|
||
f"No state transitions found in last {minutes}m.\n"
|
||
f"Device may be offline or stuck."
|
||
)
|
||
|
||
# Check for stuck state
|
||
issues = []
|
||
last_line = transitions[-1]
|
||
if "SLEEP" in last_line.split("→")[-1].strip():
|
||
issues.append(
|
||
"⚠️ Device last transitioned to SLEEP — may be stuck "
|
||
"(known FSM bug, see AGENTS.md)"
|
||
)
|
||
|
||
issues_str = "\n".join(f"- {i}" for i in issues) if issues else "None"
|
||
|
||
return (
|
||
f"## 🔄 FSM Transitions: {device_id} (last {minutes}m)\n\n"
|
||
f"```\n" + "\n".join(transitions) + "\n```\n\n"
|
||
f"**Issues:** {issues_str}"
|
||
)
|
||
|
||
except Exception as e:
|
||
return f"## ❌ FSM Analysis Failed\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="infra_fleet_health",
|
||
annotations={
|
||
"title": "Fleet Health Check",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def infra_fleet_health(minutes: int = 30) -> str:
|
||
"""Run a fleet-wide health check. Checks DynamoDB log recency for all known devices.
|
||
|
||
Args:
|
||
minutes: Lookback window in minutes (default 30).
|
||
"""
|
||
try:
|
||
import boto3
|
||
from boto3.dynamodb.conditions import Key
|
||
except ImportError:
|
||
return "Error: `boto3` not installed."
|
||
|
||
try:
|
||
dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION)
|
||
table = dynamodb.Table(DYNAMODB_TABLE)
|
||
now = datetime.now(timezone.utc)
|
||
start_ms = int((now - timedelta(minutes=minutes)).timestamp() * 1000)
|
||
|
||
healthy = []
|
||
stale = []
|
||
silent = []
|
||
|
||
for device_id in DEFAULT_DEVICES:
|
||
kce = Key("device_id").eq(device_id) & Key("timestamp").gt(
|
||
Decimal(str(start_ms))
|
||
)
|
||
resp = table.query(
|
||
KeyConditionExpression=kce,
|
||
ScanIndexForward=False,
|
||
Limit=1,
|
||
)
|
||
items = resp.get("Items", [])
|
||
|
||
if items:
|
||
latest_ts = int(items[0].get("timestamp", 0))
|
||
delta = now - datetime.fromtimestamp(
|
||
latest_ts / 1000, tz=timezone.utc
|
||
)
|
||
age_min = delta.total_seconds() / 60
|
||
if age_min < minutes:
|
||
healthy.append(f"✅ {device_id} (last log {age_min:.0f}m ago)")
|
||
else:
|
||
stale.append(f"⚠️ {device_id} (last log {age_min:.0f}m ago)")
|
||
else:
|
||
silent.append(f"❌ {device_id}")
|
||
|
||
total = len(DEFAULT_DEVICES)
|
||
return (
|
||
f"## 🌍 Fleet Health ({len(healthy)}/{total} active)\n\n"
|
||
f"### Active ({len(healthy)})\n"
|
||
+ "\n".join(f"- {h}" for h in healthy[:20])
|
||
+ ("\n- _...and more_" if len(healthy) > 20 else "")
|
||
+ f"\n\n### Stale ({len(stale)})\n"
|
||
+ ("\n".join(f"- {s}" for s in stale) if stale else "- None")
|
||
+ f"\n\n### Silent — no logs in {minutes}m ({len(silent)})\n"
|
||
+ ("\n".join(f"- {s}" for s in silent) if silent else "- None")
|
||
)
|
||
|
||
except Exception as e:
|
||
return f"## ❌ Fleet Health Check Failed\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
@mcp.tool(
|
||
name="infra_iot_things",
|
||
annotations={
|
||
"title": "List IoT Things and Principals",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def infra_iot_things(device_id: Optional[str] = None) -> str:
|
||
"""List IoT things registered in AWS, or get details for a specific device.
|
||
|
||
Args:
|
||
device_id: Specific device to inspect (optional — omit for full list).
|
||
"""
|
||
try:
|
||
import boto3
|
||
except ImportError:
|
||
return "Error: `boto3` not installed."
|
||
|
||
try:
|
||
iot = boto3.client("iot", region_name=AWS_REGION)
|
||
|
||
if device_id:
|
||
# Get thing details
|
||
try:
|
||
thing = iot.describe_thing(thingName=device_id)
|
||
except iot.exceptions.ResourceNotFoundException:
|
||
return f"## ❌ Thing `{device_id}` not found in IoT Core."
|
||
|
||
principals = iot.list_thing_principals(thingName=device_id)
|
||
cert_arns = principals.get("principals", [])
|
||
|
||
return (
|
||
f"## 🔧 IoT Thing: {device_id}\n\n"
|
||
f"- **Thing Name:** {thing.get('thingName')}\n"
|
||
f"- **Thing ID:** {thing.get('thingId')}\n"
|
||
f"- **Version:** {thing.get('version')}\n"
|
||
f"- **Certificates attached:** {len(cert_arns)}\n\n"
|
||
f"**Certificate ARNs:**\n"
|
||
+ "\n".join(f"- `{a.split('/')[-1][:12]}...`" for a in cert_arns)
|
||
)
|
||
else:
|
||
# List all things
|
||
things = []
|
||
params = {"maxResults": 100}
|
||
while True:
|
||
resp = iot.list_things(**params)
|
||
things.extend(resp.get("things", []))
|
||
if "nextToken" in resp:
|
||
params["nextToken"] = resp["nextToken"]
|
||
else:
|
||
break
|
||
|
||
lines = [f"## 🔧 IoT Things ({len(things)})"]
|
||
for t in sorted(things, key=lambda x: x.get("thingName", "")):
|
||
name = t.get("thingName", "?")
|
||
lines.append(f"- {name}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
except Exception as e:
|
||
return f"## ❌ IoT Query Failed\n\n**Error:**\n```\n{e}\n```"
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# CI TOOLS — GitHub Actions Workflow Analysis
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
# Error patterns for build failure analysis
|
||
_CI_ERROR_PATTERNS = [
|
||
{
|
||
"id": "missing_header",
|
||
"pattern": r"fatal error:\s+(\S+\.h):\s+No such file or directory",
|
||
"category": "build",
|
||
"description": "Missing header file",
|
||
"suggestion": "Add the directory containing {match} to INCLUDE_DIRS in CMakeLists.txt",
|
||
},
|
||
{
|
||
"id": "undefined_reference",
|
||
"pattern": r"undefined reference to `(\w+)'",
|
||
"category": "linker",
|
||
"description": "Undefined symbol (linker error)",
|
||
"suggestion": "Add the component containing {match} to REQUIRES in CMakeLists.txt",
|
||
},
|
||
{
|
||
"id": "test_failure",
|
||
"pattern": r"(\S+):(\d+):(\w+):FAIL:\s*(.*)",
|
||
"category": "test",
|
||
"description": "Unity test assertion failure",
|
||
"suggestion": "Test {match} failed",
|
||
},
|
||
{
|
||
"id": "coverage_threshold",
|
||
"pattern": r"Line coverage ([\d.]+)% is below ([\d.]+)% threshold",
|
||
"category": "coverage",
|
||
"description": "Coverage below threshold",
|
||
"suggestion": "Add more tests to increase line coverage from {match}",
|
||
},
|
||
{
|
||
"id": "compiler_error",
|
||
"pattern": r"(\S+\.[ch]):(\d+):\d+:\s+error:\s+(.*)",
|
||
"category": "build",
|
||
"description": "Compiler error",
|
||
"suggestion": "Fix {match}",
|
||
},
|
||
{
|
||
"id": "ninja_failed",
|
||
"pattern": r"ninja failed with exit code (\d+)",
|
||
"category": "build",
|
||
"description": "Build system failure",
|
||
"suggestion": "Check the compiler errors above for the root cause",
|
||
},
|
||
]
|
||
|
||
|
||
async def _gh_api(endpoint: str, jq_filter: str = ".") -> dict | list | str:
|
||
"""Call GitHub API via gh CLI."""
|
||
result = await _run_command(
|
||
["gh", "api", endpoint, "--jq", jq_filter], timeout=30
|
||
)
|
||
if not result["success"]:
|
||
raise RuntimeError(f"gh api failed: {result['stderr']}")
|
||
return result["stdout"]
|
||
|
||
|
||
async def _gh_run_logs(run_id: str, failed_only: bool = False) -> str:
|
||
"""Fetch run logs via gh CLI."""
|
||
args = ["gh", "run", "view", str(run_id)]
|
||
args.append("--log-failed" if failed_only else "--log")
|
||
result = await _run_command(args, timeout=120)
|
||
if not result["success"]:
|
||
if "no failed steps" in result["stderr"].lower():
|
||
return ""
|
||
raise RuntimeError(f"gh run view failed: {result['stderr']}")
|
||
return result["stdout"]
|
||
|
||
|
||
@mcp.tool(
|
||
name="ci_list_runs",
|
||
annotations={
|
||
"title": "List Recent CI Runs",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def ci_list_runs(
|
||
branch: Optional[str] = None,
|
||
limit: int = 10,
|
||
response_format: str = "markdown",
|
||
) -> str:
|
||
"""List recent GitHub Actions workflow runs with status and conclusion.
|
||
|
||
Args:
|
||
branch: Filter by branch name (optional).
|
||
limit: Number of runs to return (default 10, max 30).
|
||
response_format: Output format — 'markdown' or 'json'.
|
||
"""
|
||
limit = min(limit, 30)
|
||
endpoint = f"repos/SavearthTech/smart-device-firmware/actions/runs?per_page={limit}"
|
||
if branch:
|
||
endpoint += f"&branch={branch}"
|
||
|
||
jq = '.workflow_runs[] | {id: .id, title: .display_title, status: .status, conclusion: .conclusion, branch: .head_branch, event: .event, created: .created_at, url: .html_url}'
|
||
raw = await _gh_api(endpoint, jq)
|
||
|
||
if not raw:
|
||
return "No workflow runs found."
|
||
|
||
runs = []
|
||
for line in raw.strip().split("\n"):
|
||
if line.strip():
|
||
try:
|
||
runs.append(json.loads(line))
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
if response_format == "json":
|
||
return json.dumps(runs, indent=2)
|
||
|
||
lines = [f"## 🔄 CI Runs" + (f" (branch: `{branch}`)" if branch else "")]
|
||
lines.append("")
|
||
lines.append("| Status | ID | Branch | Title | Created |")
|
||
lines.append("|--------|-----|--------|-------|---------|")
|
||
|
||
for run in runs:
|
||
conclusion = run.get("conclusion") or run.get("status", "?")
|
||
icon = {"success": "✅", "failure": "❌", "in_progress": "🔄",
|
||
"queued": "⏳", "cancelled": "🚫"}.get(conclusion, "❓")
|
||
branch_str = run.get("branch", "?")
|
||
title = run.get("title", "?")[:50]
|
||
created = run.get("created", "?")[:19]
|
||
run_id = run.get("id", "?")
|
||
lines.append(f"| {icon} {conclusion} | `{run_id}` | `{branch_str}` | {title} | {created} |")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
@mcp.tool(
|
||
name="ci_run_status",
|
||
annotations={
|
||
"title": "CI Run Status Details",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def ci_run_status(
|
||
run_id: str,
|
||
response_format: str = "markdown",
|
||
) -> str:
|
||
"""Get detailed status of a CI workflow run including all jobs and steps.
|
||
|
||
Args:
|
||
run_id: Workflow run ID (numeric).
|
||
response_format: Output format — 'markdown' or 'json'.
|
||
"""
|
||
# Fetch run metadata
|
||
run_raw = await _gh_api(
|
||
f"repos/SavearthTech/smart-device-firmware/actions/runs/{run_id}",
|
||
'{id: .id, title: .display_title, status: .status, conclusion: .conclusion, branch: .head_branch, event: .event, created: .created_at, url: .html_url}'
|
||
)
|
||
run_data = json.loads(run_raw)
|
||
|
||
# Fetch jobs with steps
|
||
jobs_raw = await _gh_api(
|
||
f"repos/SavearthTech/smart-device-firmware/actions/runs/{run_id}/jobs",
|
||
'.jobs[] | {id: .id, name: .name, status: .status, conclusion: .conclusion, steps: [.steps[] | {name: .name, number: .number, conclusion: .conclusion}]}'
|
||
)
|
||
|
||
jobs = []
|
||
for line in jobs_raw.strip().split("\n"):
|
||
if line.strip():
|
||
try:
|
||
jobs.append(json.loads(line))
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
summary = {
|
||
"run": run_data,
|
||
"jobs": jobs,
|
||
"failed_jobs": [j["name"] for j in jobs if j.get("conclusion") == "failure"],
|
||
"failed_steps": [
|
||
{"job": j["name"], "step": s["name"]}
|
||
for j in jobs
|
||
for s in j.get("steps", [])
|
||
if s.get("conclusion") == "failure"
|
||
],
|
||
}
|
||
|
||
if response_format == "json":
|
||
return json.dumps(summary, indent=2)
|
||
|
||
conclusion = run_data.get("conclusion") or run_data.get("status", "?")
|
||
icon = {"success": "✅", "failure": "❌", "in_progress": "🔄"}.get(conclusion, "❓")
|
||
|
||
lines = [
|
||
f"## {icon} Run `{run_id}`",
|
||
f"",
|
||
f"- **Title:** {run_data.get('title', '?')}",
|
||
f"- **Status:** {conclusion}",
|
||
f"- **Branch:** `{run_data.get('branch', '?')}`",
|
||
f"- **URL:** {run_data.get('url', '?')}",
|
||
f"",
|
||
]
|
||
|
||
for job in jobs:
|
||
j_conclusion = job.get("conclusion") or job.get("status", "?")
|
||
j_icon = {"success": "✅", "failure": "❌", "skipped": "⏭️",
|
||
"in_progress": "🔄"}.get(j_conclusion, "❓")
|
||
lines.append(f"### {j_icon} {job['name']}")
|
||
|
||
for step in job.get("steps", []):
|
||
s_conclusion = step.get("conclusion") or "—"
|
||
s_icon = {"success": "✓", "failure": "✗", "skipped": "⊘"}.get(s_conclusion, "…")
|
||
lines.append(f"- {s_icon} {step['name']}")
|
||
lines.append("")
|
||
|
||
if summary["failed_steps"]:
|
||
lines.append("### ⚠️ Failed Steps")
|
||
for fs in summary["failed_steps"]:
|
||
lines.append(f"- **{fs['job']}** → {fs['step']}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
@mcp.tool(
|
||
name="ci_job_logs",
|
||
annotations={
|
||
"title": "Fetch CI Job Logs",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def ci_job_logs(
|
||
run_id: str,
|
||
job_name: Optional[str] = None,
|
||
failed_only: bool = True,
|
||
tail: int = 100,
|
||
) -> str:
|
||
"""Fetch full logs for a workflow run or specific job.
|
||
|
||
Args:
|
||
run_id: Workflow run ID.
|
||
job_name: Filter by job name substring (optional — omit for all jobs).
|
||
failed_only: If true, only fetch logs from failed steps (default: true).
|
||
tail: Number of lines to return from the end (default: 100, max: 500).
|
||
"""
|
||
tail = min(tail, 500)
|
||
|
||
logs = await _gh_run_logs(run_id, failed_only=failed_only)
|
||
|
||
if not logs:
|
||
return f"## ✅ No {'failed ' if failed_only else ''}logs for run `{run_id}`."
|
||
|
||
# Filter by job name if specified
|
||
if job_name:
|
||
filtered = []
|
||
for line in logs.split("\n"):
|
||
if job_name.lower() in line.lower():
|
||
filtered.append(line)
|
||
logs = "\n".join(filtered)
|
||
|
||
# Strip ANSI codes for clean output
|
||
logs = re.sub(r"\033\[[0-9;]*[mK]", "", logs)
|
||
|
||
# Tail
|
||
log_lines = logs.strip().split("\n")
|
||
total = len(log_lines)
|
||
if total > tail:
|
||
log_lines = log_lines[-tail:]
|
||
truncation_msg = f"\n> Showing last {tail} of {total} lines. Use `tail` parameter for more.\n\n"
|
||
else:
|
||
truncation_msg = ""
|
||
|
||
return (
|
||
f"## 📋 CI Logs — Run `{run_id}`"
|
||
+ (f" (job: `{job_name}`)" if job_name else "")
|
||
+ f"\n\n{truncation_msg}"
|
||
+ f"```\n" + "\n".join(log_lines) + "\n```"
|
||
)
|
||
|
||
|
||
@mcp.tool(
|
||
name="ci_analyze_failure",
|
||
annotations={
|
||
"title": "Analyze CI Failure",
|
||
"readOnlyHint": True,
|
||
"destructiveHint": False,
|
||
"idempotentHint": True,
|
||
"openWorldHint": True,
|
||
},
|
||
)
|
||
async def ci_analyze_failure(
|
||
run_id: str,
|
||
response_format: str = "markdown",
|
||
) -> str:
|
||
"""Analyze a failed CI run: extract errors, identify patterns, and suggest fixes.
|
||
|
||
Detects: missing headers, linker errors, test failures, coverage threshold
|
||
violations, and compiler errors. For missing headers, searches the workspace
|
||
to locate the file and suggest the correct include path.
|
||
|
||
Args:
|
||
run_id: Workflow run ID.
|
||
response_format: Output format — 'markdown' or 'json'.
|
||
"""
|
||
# Get failed logs
|
||
logs = await _gh_run_logs(run_id, failed_only=True)
|
||
|
||
if not logs:
|
||
return f"## ✅ No failures in run `{run_id}`."
|
||
|
||
# Strip ANSI codes
|
||
clean_logs = re.sub(r"\033\[[0-9;]*[mK]", "", logs)
|
||
log_lines = clean_logs.split("\n")
|
||
|
||
errors = []
|
||
for i, line in enumerate(log_lines):
|
||
for pat in _CI_ERROR_PATTERNS:
|
||
m = re.search(pat["pattern"], line)
|
||
if m:
|
||
# Context
|
||
ctx_before = log_lines[max(0, i - 2):i]
|
||
ctx_after = log_lines[i + 1:min(len(log_lines), i + 3)]
|
||
|
||
# Job/step from tab-delimited prefix
|
||
parts = line.split("\t", 2)
|
||
job_name = parts[0].strip() if len(parts) >= 1 else ""
|
||
step_name = parts[1].strip() if len(parts) >= 2 else ""
|
||
|
||
match_str = m.group(1) if m.lastindex and m.lastindex >= 1 else m.group(0)
|
||
suggestion = pat["suggestion"].format(match=match_str)
|
||
|
||
errors.append({
|
||
"pattern_id": pat["id"],
|
||
"category": pat["category"],
|
||
"description": pat["description"],
|
||
"matched_text": m.group(0),
|
||
"match_key": match_str,
|
||
"suggestion": suggestion,
|
||
"job": job_name,
|
||
"step": step_name,
|
||
"line": i + 1,
|
||
"context_before": ctx_before,
|
||
"context_after": ctx_after,
|
||
})
|
||
|
||
# Deduplicate
|
||
seen = set()
|
||
unique = []
|
||
for err in errors:
|
||
key = (err["pattern_id"], err["matched_text"])
|
||
if key not in seen:
|
||
seen.add(key)
|
||
unique.append(err)
|
||
|
||
# For missing headers, locate them in workspace
|
||
header_locations = {}
|
||
for err in unique:
|
||
if err["pattern_id"] == "missing_header":
|
||
header = re.search(r"fatal error:\s+(\S+\.h)", err["matched_text"])
|
||
if header:
|
||
h_name = header.group(1)
|
||
result = await _run_command(
|
||
["find", PROJECT_DIR, "-name", h_name,
|
||
"-not", "-path", "*/build/*", "-not", "-path", "*/.git/*"],
|
||
timeout=10,
|
||
)
|
||
if result["success"] and result["stdout"].strip():
|
||
locations = result["stdout"].strip().split("\n")
|
||
header_locations[h_name] = locations
|
||
|
||
result_data = {
|
||
"run_id": run_id,
|
||
"total_matches": len(errors),
|
||
"unique_errors": len(unique),
|
||
"errors": unique,
|
||
"header_locations": header_locations,
|
||
"categories": {
|
||
cat: sum(1 for e in unique if e["category"] == cat)
|
||
for cat in set(e["category"] for e in unique)
|
||
},
|
||
}
|
||
|
||
if response_format == "json":
|
||
return json.dumps(result_data, indent=2)
|
||
|
||
if not unique:
|
||
return (
|
||
f"## ⚠️ Run `{run_id}` failed but no recognized error patterns found.\n\n"
|
||
f"Use `ci_job_logs` to view the raw failed logs."
|
||
)
|
||
|
||
lines = [
|
||
f"## 🔍 CI Failure Analysis — Run `{run_id}`",
|
||
f"",
|
||
f"Found **{len(unique)}** unique error(s)",
|
||
f"",
|
||
]
|
||
|
||
for idx, err in enumerate(unique, 1):
|
||
lines.append(f"### Error #{idx}: {err['description']}")
|
||
if err.get("job"):
|
||
lines.append(f"- **Job:** {err['job']}")
|
||
if err.get("step"):
|
||
lines.append(f"- **Step:** {err['step']}")
|
||
lines.append(f"- **Match:** `{err['matched_text']}`")
|
||
lines.append(f"- **Fix:** {err['suggestion']}")
|
||
|
||
# Show context
|
||
if err.get("context_before") or err.get("context_after"):
|
||
lines.append(f"")
|
||
lines.append("```")
|
||
for ctx in err["context_before"][-2:]:
|
||
ctx_text = ctx.split("\t")[-1] if "\t" in ctx else ctx
|
||
lines.append(ctx_text.rstrip())
|
||
lines.append(f">>> {err['matched_text']}")
|
||
for ctx in err["context_after"][:2]:
|
||
ctx_text = ctx.split("\t")[-1] if "\t" in ctx else ctx
|
||
lines.append(ctx_text.rstrip())
|
||
lines.append("```")
|
||
lines.append("")
|
||
|
||
# Missing header locations
|
||
if header_locations:
|
||
lines.append("### 📍 Header Locations Found")
|
||
for h_name, locs in header_locations.items():
|
||
for loc in locs[:3]:
|
||
rel = os.path.relpath(loc, PROJECT_DIR)
|
||
parent = os.path.dirname(rel)
|
||
lines.append(f"- `{h_name}` → `{rel}`")
|
||
lines.append(f" - Add to INCLUDE_DIRS: `\"../../../{parent}\"`")
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
@mcp.tool()
|
||
async def memory_search(
|
||
query: str,
|
||
response_format: ResponseFormat = ResponseFormat.MARKDOWN,
|
||
) -> str:
|
||
"""
|
||
Search the persistent Hybrid Memory Engine (replica-omnisciente/.aurelio/memory/) for knowledge graph entities.
|
||
|
||
Args:
|
||
query: The topic or keywords to search for.
|
||
response_format: Output format — 'markdown' or 'json'.
|
||
"""
|
||
memory_dir = Path(PROJECT_DIR) / "replica-omnisciente" / ".aurelio" / "memory"
|
||
|
||
if not memory_dir.exists():
|
||
return f"Error: Memory engine directory not found at {memory_dir}"
|
||
|
||
results = []
|
||
terms = [t.lower() for t in query.split() if t.strip()]
|
||
|
||
if not terms:
|
||
return "Error: Empty query."
|
||
|
||
for md_file in memory_dir.rglob("*.md"):
|
||
try:
|
||
content = md_file.read_text(encoding="utf-8")
|
||
lower_content = content.lower()
|
||
|
||
score = 0
|
||
for term in terms:
|
||
# Basic term frequency scoring
|
||
score += lower_content.count(term)
|
||
|
||
if score > 0:
|
||
rel_path = md_file.relative_to(memory_dir)
|
||
snippet = ""
|
||
for line in content.split('\n'):
|
||
if any(term in line.lower() for term in terms):
|
||
snippet = line.strip()[:150]
|
||
break
|
||
|
||
results.append({
|
||
"file": str(rel_path),
|
||
"score": score,
|
||
"snippet": snippet
|
||
})
|
||
except Exception:
|
||
pass
|
||
|
||
results.sort(key=lambda x: x["score"], reverse=True)
|
||
|
||
if response_format == ResponseFormat.JSON:
|
||
return json.dumps({"query": query, "results": results[:10]}, indent=2)
|
||
|
||
if not results:
|
||
return f"No results found in memory graph for query: '{query}'"
|
||
|
||
lines = [f"## 🧠 Memory Search Results for '{query}'\n"]
|
||
for i, res in enumerate(results[:10], 1):
|
||
lines.append(f"{i}. **{res['file']}** (Score: {res['score']})")
|
||
lines.append(f" > {res['snippet']}...")
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
# ENTRY POINT — supports stdio and SSE transports
|
||
# ═════════════════════════════════════════════════════════════════════
|
||
|
||
|
||
def main():
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="savearth Unified MCP Server")
|
||
parser.add_argument(
|
||
"--transport",
|
||
choices=["stdio", "sse", "streamable-http"],
|
||
default="stdio",
|
||
help="Transport mode (default: stdio)",
|
||
)
|
||
parser.add_argument(
|
||
"--port",
|
||
type=int,
|
||
default=int(os.environ.get("MCP_PORT", "8080")),
|
||
help="HTTP port for SSE transport (default: 8080)",
|
||
)
|
||
parser.add_argument(
|
||
"--host",
|
||
default=os.environ.get("MCP_HOST", "0.0.0.0"),
|
||
help="Bind host for SSE transport (default: 0.0.0.0)",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
# FastMCP v1.27+: host/port are constructor settings, not run() kwargs.
|
||
# Update the settings object before calling run().
|
||
if args.transport in ("sse", "streamable-http"):
|
||
mcp.settings.host = args.host
|
||
mcp.settings.port = args.port
|
||
|
||
mcp.run(transport=args.transport)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|