Improve external service discovery and tool routing

platform/mcp/platform_context_mcp.py

@@ -15,15 +15,18 @@ import calendar
 import json
 import os
 import re
+import socket
 import sys
 import tempfile
 import time
 import uuid
+import urllib.error
+import urllib.request
 from pathlib import Path
 from typing import Any
 
 
-SERVER_VERSION = "1.0.0"
+SERVER_VERSION = "1.1.0"
 REPO_ROOT = Path(os.environ.get("ATHENA_REPO_ROOT", "/knowledge/repo"))
 DOCS_ROOT = Path(os.environ.get("ATHENA_DOCS_ROOT", "/workspace/docs"))
 RUNTIME_FILE = Path(os.environ.get("ATHENA_RUNTIME_FILE", "/runtime/runtime.json"))
@@ -64,6 +67,18 @@ TOOLS = [
         ),
         "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
     },
+    {
+        "name": "athena_get_external_services",
+        "description": (
+            "USE before designing or installing an integration that may already run on Unraid "
+            "or elsewhere in the home network. Returns the versioned, secret-free service "
+            "catalog and performs only fixed bounded reachability checks for those catalogued "
+            "endpoints. It accepts no host, URL or port from the model and is not a scanner. "
+            "A failed check means unavailable or unverified; it never authorizes creating a "
+            "duplicate service. Use the listed specialist MCP for detailed current state."
+        ),
+        "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
+    },
     {
         "name": "athena_search_knowledge",
         "description": (
@@ -293,6 +308,63 @@ def current_state() -> dict[str, Any]:
     return data
 
 
+def external_services() -> dict[str, Any]:
+    catalog_path = REPO_ROOT / "config/service-catalog.json"
+    try:
+        catalog = json.loads(catalog_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        return {
+            "available": False,
+            "error": str(exc),
+            "instruction": "The service inventory is unavailable. Do not infer that a replacement service is needed.",
+        }
+
+    results = []
+    for service in catalog.get("services", [])[:24]:
+        item = {key: service.get(key) for key in (
+            "id", "name", "location", "host", "address", "port", "protocol",
+            "specialist_tool", "purpose",
+        )}
+        address = str(service.get("address", ""))
+        port = int(service.get("port", 0))
+        started = time.monotonic()
+        reachable = False
+        http_status = None
+        error = None
+        try:
+            with socket.create_connection((address, port), timeout=2):
+                reachable = True
+            if service.get("probe") == "http-head":
+                path = str(service.get("probe_path", "/"))
+                url = f"{service.get('protocol', 'http')}://{address}:{port}{path}"
+                request = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "MikeAI-Service-Catalog/1"})
+                try:
+                    with urllib.request.urlopen(request, timeout=3) as response:
+                        http_status = response.status
+                except urllib.error.HTTPError as exc:
+                    http_status = exc.code
+        except (OSError, ValueError, urllib.error.URLError) as exc:
+            error = type(exc).__name__
+        item["check"] = {
+            "reachable": reachable,
+            "http_status": http_status,
+            "elapsed_ms": round((time.monotonic() - started) * 1000),
+            "error_class": error,
+        }
+        results.append(item)
+    return {
+        "available": True,
+        "source": "config/service-catalog.json",
+        "catalog_version": catalog.get("version"),
+        "updated": catalog.get("updated"),
+        "services": results,
+        "instruction": (
+            "Existing catalog entries are architecture constraints, not disposable suggestions. "
+            "If a check or specialist tool fails, report the gap and ask for direction; do not plan a duplicate backend."
+        ),
+    }
+
+
 def candidate_files() -> list[Path]:
     files: list[Path] = []
     for path in REPO_ROOT.rglob("*"):
@@ -550,6 +622,8 @@ def call_tool(name: str, arguments: dict[str, Any]) -> str:
         result = overview()
     elif name == "athena_get_current_state":
         result = current_state()
+    elif name == "athena_get_external_services":
+        result = external_services()
     elif name == "athena_search_knowledge":
         result = search_knowledge(arguments)
     elif name == "athena_read_source":