Pouzor
7935b671d3
Exposes homelab topology to MCP-compatible AI clients (Claude Code, etc.) over LAN via HTTP/SSE on port 8001. - New mcp/ service: FastAPI + mcp SDK, SSE transport - Auth: X-API-Key for AI clients, X-MCP-Service-Key for backend (Docker-internal) - Resources: canvas, nodes, edges, scan/pending, scan/runs - Tools: create/update/delete nodes+edges, trigger scan, approve/hide devices - Backend deps.py: accepts JWT or MCP service key (no plain-text password) - 40 tests (auth, resources, tools) across asyncio + trio - docker-compose.yml: mcp service on port 8001 - README: MCP setup section with Claude Code/Desktop config examples
44 lines
1.8 KiB
Python
44 lines
1.8 KiB
Python
import json
|
|
from mcp.server import Server
|
|
from mcp.types import Resource, TextContent
|
|
from .backend_client import backend
|
|
|
|
RESOURCE_LIST = [
|
|
Resource(uri="homelable://canvas", name="Canvas", description="Full canvas state (nodes + edges + viewport)", mimeType="application/json"),
|
|
Resource(uri="homelable://nodes", name="Nodes", description="All nodes in the homelab", mimeType="application/json"),
|
|
Resource(uri="homelable://edges", name="Edges", description="All network edges/links", mimeType="application/json"),
|
|
Resource(uri="homelable://scan/pending", name="Pending devices", description="Discovered devices awaiting approval", mimeType="application/json"),
|
|
Resource(uri="homelable://scan/runs", name="Scan history", description="Recent scan run history", mimeType="application/json"),
|
|
]
|
|
|
|
ROUTES = {
|
|
"homelable://canvas": "/api/v1/canvas",
|
|
"homelable://nodes": "/api/v1/nodes",
|
|
"homelable://edges": "/api/v1/edges",
|
|
"homelable://scan/pending": "/api/v1/scan/pending",
|
|
"homelable://scan/runs": "/api/v1/scan/runs",
|
|
}
|
|
|
|
|
|
async def read_resource(uri: str) -> list[TextContent]:
|
|
if uri.startswith("homelable://nodes/") and uri != "homelable://nodes/":
|
|
node_id = uri.split("/")[-1]
|
|
data = await backend.get(f"/api/v1/nodes/{node_id}")
|
|
return [TextContent(type="text", text=json.dumps(data, indent=2))]
|
|
|
|
if uri not in ROUTES:
|
|
raise ValueError(f"Unknown resource URI: {uri}")
|
|
|
|
data = await backend.get(ROUTES[uri])
|
|
return [TextContent(type="text", text=json.dumps(data, indent=2))]
|
|
|
|
|
|
def register_resources(server: Server):
|
|
@server.list_resources()
|
|
async def _list():
|
|
return RESOURCE_LIST
|
|
|
|
@server.read_resource()
|
|
async def _read(uri: str):
|
|
return await read_resource(uri)
|