feat(standalone): support Postgres via DB_ENGINE env var

Lets a standalone install point at an existing Postgres database
(e.g. for testing against real data on a verification host) instead
of always writing to local SQLite. Defaults to SQLite when DB_ENGINE
is unset, so existing self-hosters see no behavior change.

.gitignore

@@ -0,0 +1,27 @@
+# Secrets — setup.sh writes Bambu credentials + Django secret key here
+.env
+!.env.example
+
+# Local-only agent notes
+CLAUDE.local.md
+.claude/
+
+# Native setup artifacts
+.venv/
+data/
+db.sqlite3
+staticfiles/
+
+# Python
+__pycache__/
+*.py[cod]
+*.egg-info/
+build/
+dist/
+
+# Editors / OS
+.idea/
+.vscode/
+.DS_Store
+
+.worktrees/

README.md

@@ -63,6 +63,7 @@ It runs quietly in the background 24/7, capturing every print, filament change,
   - [Troubleshooting (Native)](#troubleshooting-native)
 - [Docker Setup](#docker-setup)
 - [Batch Importing Filament Colors and Filament Types](#batch-importing-filament-colors-and-filament-types)
+- [MCP Server (AI Agent Access)](#mcp-server-ai-agent-access)
 
 ---
 
@@ -296,3 +297,37 @@ Processing: PLA Basic.txt  →  type='PLA'  sub_type='PLA Basic'
 - **Created** — new color entries added to the database
 - **Skipped (duplicate)** — already existed, not changed
 - **Skipped (no type)** — only shown if `--no-auto-create-filament-type` is used and the filament type isn't in the database yet
+
+## MCP Server (AI Agent Access)
+
+Bambu-Run exposes printer status, print history, and filament inventory as
+read-only [MCP](https://modelcontextprotocol.io/) tools, so an AI agent (e.g.
+Claude Desktop, Claude Code) can query your printer directly.
+
+Install the extra:
+
+```bash
+pip install 'bambu-run[mcp]'
+```
+
+**stdio** (for a local agent on the same machine — e.g. Claude Desktop):
+
+```bash
+python manage.py bambu_mcp_server --transport stdio
+```
+
+**HTTP** (for a remote agent, or multiple callers over the network):
+
+```bash
+export BAMBU_RUN_MCP_API_KEY=your-secret-key
+python manage.py bambu_mcp_server --transport http --host 0.0.0.0 --port 8808
+```
+
+`BAMBU_RUN_MCP_API_KEY` is **required** for HTTP — the server refuses to start
+without it, since an unauthenticated network MCP endpoint would expose your
+printer and filament data to anyone who can reach the port. stdio needs no key
+(the OS process boundary is the trust boundary there).
+
+If serving behind a reverse proxy on a real domain, pass
+`--allowed-host your.domain.com` (repeatable) — otherwise the SDK's
+DNS-rebinding guard rejects requests with a real `Host` header.

bambu_run/conf.py

@@ -64,14 +64,6 @@ class _Settings:
     def MCP_PORT(self):
         return get_setting("BAMBU_RUN_MCP_PORT", 8808)
 
-    @property
-    def MCP_AUTH_BACKEND(self):
-        return get_setting("BAMBU_RUN_MCP_AUTH_BACKEND", None)
-
-    @property
-    def MCP_HIDE_SENSITIVE(self):
-        return get_setting("BAMBU_RUN_MCP_HIDE_SENSITIVE", False)
-
     # Cloud sync settings
     @property
     def CLOUD_SYNC_ENABLED(self):

bambu_run/management/commands/bambu_mcp_server.py

@@ -1,355 +1,110 @@
-"""
-Management command to run the Bambu-Run MCP server.
-
-Supports SSE (network) and stdio (local) transports.
+"""Run the Bambu-Run MCP server (standalone, single-owner).
 
 Usage:
-    python manage.py bambu_mcp_server
-    python manage.py bambu_mcp_server --transport sse --host 0.0.0.0 --port 8808
     python manage.py bambu_mcp_server --transport stdio
+    python manage.py bambu_mcp_server --transport http --host 0.0.0.0 --port 8808
+    python manage.py bambu_mcp_server --transport http --allowed-host mcp.example.com
 """
-
 import logging
 
 from django.core.management.base import BaseCommand, CommandError
 
 logger = logging.getLogger("bambu_run.mcp")
 
+ALL_CAPABILITIES = ["read_printer", "read_print_history", "read_filament"]
+
+_INSTRUCTIONS = (
+    "Read-only access to Bambu Lab 3D printer data: live status, print "
+    "history, and filament inventory, collected via MQTT."
+)
+
+
+def _build_gate(kit, transport, api_key=None):
+    """Build the Gate for the requested transport.
+
+    Raises ValueError if HTTP is requested without an api_key -- an
+    unauthenticated network MCP endpoint would expose printer and filament
+    data to anyone who can reach the port. stdio always uses OpenIdentity:
+    the OS process boundary is the trust boundary there, matching mcp-gate's
+    own guidance (never use OpenIdentity over HTTP).
+    """
+    from mcp_gate import Gate, OpenIdentity, StaticTokenIdentity
+
+    if transport == "stdio":
+        return Gate(name="bambu-run", toolkits=[kit], identity=OpenIdentity(ALL_CAPABILITIES), instructions=_INSTRUCTIONS)
+
+    if not api_key:
+        raise ValueError(
+            "BAMBU_RUN_MCP_API_KEY must be set to serve MCP over HTTP -- an "
+            "unauthenticated network MCP endpoint would expose printer and "
+            "filament data to anyone who can reach the port."
+        )
+    return Gate(
+        name="bambu-run",
+        toolkits=[kit],
+        identity=StaticTokenIdentity({api_key: ("owner", ALL_CAPABILITIES)}),
+        instructions=_INSTRUCTIONS,
+    )
+
 
 class Command(BaseCommand):
-    help = "Run the Bambu-Run MCP server for AI agent access"
+    help = "Run the Bambu-Run MCP server for AI agent access (standalone, single-owner)"
 
     def add_arguments(self, parser):
         from bambu_run.conf import app_settings
 
+        parser.add_argument("--transport", choices=["http", "stdio"], default="stdio")
+        parser.add_argument("--host", default=app_settings.MCP_HOST)
+        parser.add_argument("--port", type=int, default=app_settings.MCP_PORT)
         parser.add_argument(
-            "--transport",
-            choices=["sse", "stdio"],
-            default="sse",
-            help="Transport mode (default: sse)",
-        )
-        parser.add_argument(
-            "--host",
-            default=app_settings.MCP_HOST,
-            help=f"Host to bind to (default: {app_settings.MCP_HOST})",
-        )
-        parser.add_argument(
-            "--port",
-            type=int,
-            default=app_settings.MCP_PORT,
-            help=f"Port to listen on (default: {app_settings.MCP_PORT})",
+            "--allowed-host",
+            action="append",
+            default=None,
+            help=(
+                "Host header this server should accept (repeatable). Required "
+                "when reachable via a real domain or reverse proxy, e.g. "
+                "--allowed-host mcp.example.com. Omit for local-only testing "
+                "on 127.0.0.1/localhost."
+            ),
         )
 
     def handle(self, *args, **options):
         try:
-            from mcp.server.fastmcp import FastMCP
+            from mcp_gate import run_stdio, asgi_app
         except ImportError:
             raise CommandError(
-                "The 'mcp' package is required. Install it with: pip install 'bambu-run[mcp]'"
+                "The 'mcp-gate' package is required. Install it with: pip install 'bambu-run[mcp]'"
             )
 
-        from asgiref.sync import sync_to_async
         from bambu_run.conf import app_settings
-        from bambu_run import mcp_tools
+        from bambu_run.mcp.toolkit import build_toolkit
 
         transport = options["transport"]
-        host = options["host"]
-        port = options["port"]
+        kit = build_toolkit()
 
-        mcp = FastMCP(
-            "Bambu-Run",
-            instructions=(
-                "Bambu-Run MCP server provides read-only access to 3D printer data "
-                "including live printer status, filament inventory, print history, "
-                "temperature trends, and diagnostics. All data comes from Bambu Lab "
-                "printers monitored via MQTT."
-            ),
-        )
+        if transport == "stdio":
+            gate = _build_gate(kit, transport="stdio")
+            self.stdout.write(self.style.SUCCESS("Starting Bambu-Run MCP server (stdio)"))
+            run_stdio(gate)
+            return
 
-        # ── Register Tools ───────────────────────────────────────────────
+        try:
+            gate = _build_gate(kit, transport="http", api_key=app_settings.MCP_API_KEY)
+        except ValueError as exc:
+            raise CommandError(str(exc))
 
-        @mcp.tool()
-        async def get_printer_status(printer_id: int | None = None) -> str:
-            """Get current live status of printer(s) including temperatures, progress, AMS slots, and errors.
+        try:
+            import uvicorn
+        except ImportError:
+            raise CommandError("uvicorn is required for HTTP transport. Install it with: pip install uvicorn")
 
-            Args:
-                printer_id: Optional printer ID to filter. Omit for all printers.
-            """
-            return await sync_to_async(mcp_tools.get_printer_status)(printer_id=printer_id)
+        asgi_kwargs = {}
+        if options["allowed_host"]:
+            from mcp.server.transport_security import TransportSecuritySettings
+            asgi_kwargs["transport_security"] = TransportSecuritySettings(allowed_hosts=options["allowed_host"])
 
-        @mcp.tool()
-        async def list_printers() -> str:
-            """List all registered printers with their model, serial, IP, and active status."""
-            return await sync_to_async(mcp_tools.list_printers)()
-
-        @mcp.tool()
-        async def get_print_history(
-            status: str | None = None,
-            days: int | None = None,
-            project_name: str | None = None,
-            limit: int = 20,
-        ) -> str:
-            """Get print job history with optional filters.
-
-            Args:
-                status: Filter by status (FINISH, FAILED, CANCELLED).
-                days: Only show jobs from the last N days.
-                project_name: Filter by project name (partial match).
-                limit: Maximum number of results (default 20).
-            """
-            return await sync_to_async(mcp_tools.get_print_history)(
-                status=status, days=days, project_name=project_name, limit=limit
-            )
-
-        @mcp.tool()
-        async def get_print_job_detail(job_id: int) -> str:
-            """Get detailed information about a single print job including filament usage.
-
-            Args:
-                job_id: The print job ID.
-            """
-            return await sync_to_async(mcp_tools.get_print_job_detail)(job_id=job_id)
-
-        @mcp.tool()
-        async def list_filaments(
-            type: str | None = None,
-            brand: str | None = None,
-            color: str | None = None,
-            loaded_in_ams: bool | None = None,
-            low_filament: bool | None = None,
-        ) -> str:
-            """List filament inventory with optional filters.
-
-            Args:
-                type: Filter by material type (PLA, PETG, ABS, etc.).
-                brand: Filter by brand name (partial match).
-                color: Filter by color name (partial match).
-                loaded_in_ams: Filter by whether spool is currently in AMS.
-                low_filament: If true, only show spools with <=20% remaining.
-            """
-            return await sync_to_async(mcp_tools.list_filaments)(
-                type=type, brand=brand, color=color,
-                loaded_in_ams=loaded_in_ams, low_filament=low_filament,
-            )
-
-        @mcp.tool()
-        async def get_filament_detail(filament_id: int) -> str:
-            """Get detailed information about a single filament spool including usage history.
-
-            Args:
-                filament_id: The filament spool ID.
-            """
-            return await sync_to_async(mcp_tools.get_filament_detail)(filament_id=filament_id)
-
-        @mcp.tool()
-        async def get_temperature_history(
-            printer_id: int | None = None,
-            hours: int = 6,
-            metric: str = "all",
-        ) -> str:
-            """Get temperature trends (avg/min/max) over recent hours.
-
-            Args:
-                printer_id: Optional printer ID to filter.
-                hours: Number of hours to look back (default 6).
-                metric: Which sensor to show: 'all', 'nozzle', 'bed', or 'chamber'.
-            """
-            return await sync_to_async(mcp_tools.get_temperature_history)(
-                printer_id=printer_id, hours=hours, metric=metric
-            )
-
-        @mcp.tool()
-        async def get_filament_usage_stats(days: int = 30, group_by: str = "type") -> str:
-            """Get aggregate filament consumption statistics.
-
-            Args:
-                days: Number of days to look back (default 30).
-                group_by: Group results by 'type', 'color', or 'spool'.
-            """
-            return await sync_to_async(mcp_tools.get_filament_usage_stats)(days=days, group_by=group_by)
-
-        @mcp.tool()
-        async def get_printer_health(printer_id: int | None = None) -> str:
-            """Get printer diagnostics including errors, humidity, WiFi signal, and recent failures.
-
-            Args:
-                printer_id: Optional printer ID to filter. Omit for all printers.
-            """
-            return await sync_to_async(mcp_tools.get_printer_health)(printer_id=printer_id)
-
-        @mcp.tool()
-        async def search_print_jobs(query: str) -> str:
-            """Search print jobs by project name or gcode filename.
-
-            Args:
-                query: Search text (partial match on project name or gcode file).
-            """
-            return await sync_to_async(mcp_tools.search_print_jobs)(query=query)
-
-        @mcp.tool()
-        async def get_printing_summary(days: int = 7) -> str:
-            """Get high-level printing activity summary including job counts, success rate, and top projects.
-
-            Args:
-                days: Number of days to summarize (default 7).
-            """
-            return await sync_to_async(mcp_tools.get_printing_summary)(days=days)
-
-        @mcp.tool()
-        async def find_compatible_filament(
-            type: str,
-            min_remaining_percent: int = 10,
-            color: str | None = None,
-        ) -> str:
-            """Find filament spools matching material type and optional criteria.
-
-            Args:
-                type: Material type to search for (PLA, PETG, ABS, etc.).
-                min_remaining_percent: Minimum remaining percentage (default 10).
-                color: Optional color filter (partial match).
-            """
-            return await sync_to_async(mcp_tools.find_compatible_filament)(
-                type=type, min_remaining_percent=min_remaining_percent, color=color
-            )
-
-        # ── Register Resources ───────────────────────────────────────────
-
-        @mcp.resource("bambu://printers")
-        async def res_printers() -> str:
-            """List all registered printers."""
-            return await sync_to_async(mcp_tools.resource_printers)()
-
-        @mcp.resource("bambu://printers/{printer_id}/status")
-        async def res_printer_status(printer_id: int) -> str:
-            """Get latest status for a specific printer."""
-            return await sync_to_async(mcp_tools.resource_printer_status)(printer_id)
-
-        @mcp.resource("bambu://filaments")
-        async def res_filaments() -> str:
-            """Full filament inventory."""
-            return await sync_to_async(mcp_tools.resource_filaments)()
-
-        @mcp.resource("bambu://filaments/{filament_id}")
-        async def res_filament_detail(filament_id: int) -> str:
-            """Single filament spool with usage history."""
-            return await sync_to_async(mcp_tools.resource_filament_detail)(filament_id)
-
-        @mcp.resource("bambu://print-jobs/recent")
-        async def res_recent_jobs() -> str:
-            """Last 20 print jobs."""
-            return await sync_to_async(mcp_tools.resource_recent_print_jobs)()
-
-        @mcp.resource("bambu://filament-types")
-        async def res_filament_types() -> str:
-            """Filament type registry."""
-            return await sync_to_async(mcp_tools.resource_filament_types)()
-
-        @mcp.resource("bambu://filament-colors")
-        async def res_filament_colors() -> str:
-            """Filament color database."""
-            return await sync_to_async(mcp_tools.resource_filament_colors)()
-
-        # ── Register Prompts ─────────────────────────────────────────────
-
-        @mcp.prompt()
-        async def printer_check_in(printer_id: int | None = None) -> str:
-            """Full printer status briefing with health check and recent prints.
-
-            Args:
-                printer_id: Optional printer ID. Omit for all printers.
-            """
-            return await sync_to_async(mcp_tools.prompt_printer_check_in)(printer_id=printer_id)
-
-        @mcp.prompt()
-        async def filament_inventory_report() -> str:
-            """Comprehensive filament inventory report with low-stock warnings."""
-            return await sync_to_async(mcp_tools.prompt_filament_inventory_report)()
-
-        @mcp.prompt()
-        async def print_job_review(job_id: int) -> str:
-            """Detailed review of a completed print job.
-
-            Args:
-                job_id: The print job ID to review.
-            """
-            return await sync_to_async(mcp_tools.prompt_print_job_review)(job_id)
-
-        @mcp.prompt()
-        async def weekly_printing_digest() -> str:
-            """Weekly printing activity summary with filament usage breakdown."""
-            return await sync_to_async(mcp_tools.prompt_weekly_digest)()
-
-        @mcp.prompt()
-        async def troubleshoot_printer(printer_id: int | None = None) -> str:
-            """Diagnose printer issues using recent health data, status, and temperatures.
-
-            Args:
-                printer_id: Optional printer ID. Omit for all printers.
-            """
-            return await sync_to_async(mcp_tools.prompt_troubleshoot_printer)(printer_id=printer_id)
-
-        # ── Auth middleware for SSE ───────────────────────────────────────
-
-        api_key = app_settings.MCP_API_KEY
-        auth_backend = app_settings.MCP_AUTH_BACKEND
-
-        if api_key or auth_backend:
-            from starlette.middleware.base import BaseHTTPMiddleware
-            from starlette.responses import JSONResponse
-
-            class AuthMiddleware(BaseHTTPMiddleware):
-                async def dispatch(self, request, call_next):
-                    # Custom auth backend takes priority
-                    if auth_backend:
-                        if not auth_backend(request):
-                            return JSONResponse(
-                                {"error": "Unauthorized"}, status_code=401
-                            )
-                        return await call_next(request)
-
-                    # API key auth
-                    if api_key:
-                        auth_header = request.headers.get("Authorization", "")
-                        if auth_header == f"Bearer {api_key}":
-                            return await call_next(request)
-                        return JSONResponse(
-                            {"error": "Invalid or missing API key"}, status_code=401
-                        )
-
-                    return await call_next(request)
-
-            # Attach middleware — FastMCP's SSE app is a Starlette app
-            original_sse_app = mcp.sse_app
-
-            def patched_sse_app():
-                app = original_sse_app()
-                app.add_middleware(AuthMiddleware)
-                return app
-
-            mcp.sse_app = patched_sse_app
-
-        # ── Run ──────────────────────────────────────────────────────────
-
-        if transport == "sse":
-            try:
-                import uvicorn
-            except ImportError:
-                raise CommandError(
-                    "uvicorn is required for SSE transport. Install it with: pip install uvicorn"
-                )
-
-            self.stdout.write(
-                self.style.SUCCESS(
-                    f"Starting Bambu-Run MCP server (SSE) on {host}:{port}"
-                )
-            )
-            self.stdout.write(
-                f"Connect with: http://{host}:{port}/sse"
-            )
-            app = mcp.sse_app()
-            uvicorn.run(app, host=host, port=port)
-        else:
-            self.stdout.write(
-                self.style.SUCCESS("Starting Bambu-Run MCP server (stdio)")
-            )
-            mcp.run(transport="stdio")
+        app = asgi_app(gate, **asgi_kwargs)
+        host, port = options["host"], options["port"]
+        self.stdout.write(self.style.SUCCESS(f"Starting Bambu-Run MCP server (HTTP) on {host}:{port}"))
+        self.stdout.write(f"Connect with: http://{host}:{port}/mcp  (Authorization: Bearer <key>)")
+        uvicorn.run(app, host=host, port=port)

bambu_run/mcp/__init__.py

@@ -0,0 +1 @@
+"""MCP toolkit for Bambu-Run: read-only printer, print-history, and filament tools."""

bambu_run/mcp/filaments.py

@@ -0,0 +1,118 @@
+"""list_filaments, get_filament -- the read_filament capability tools."""
+from __future__ import annotations
+
+from django.utils import timezone
+
+from bambu_run.mcp.shapes import normalize_consumed, remaining_status
+from bambu_run.models import Filament
+
+
+def list_filaments(
+    material: str | None = None,
+    sub_type: str | None = None,
+    brand: str | None = None,
+    color: str | None = None,
+    loaded_only: bool = False,
+    min_remaining_percent: int | None = None,
+    min_remaining_grams: int | None = None,
+    include_history: bool = False,
+    limit: int = 50,
+) -> dict:
+    """Filament spool inventory with composable filters.
+
+    "PLA Matte Black" is material="PLA", sub_type="PLA Matte", color="black".
+    "250g of black PLA" is material="PLA", color="black",
+    min_remaining_grams=250. remaining_percent/remaining_grams are null when
+    the spool's RFID has never been read (remaining_status is "unread" in
+    that case, not "low" or "empty"). include_history (default off) adds
+    each spool's last 3 uses inline; call get_filament for the complete
+    history of one spool. limit hard ceiling is 100.
+    """
+    limit = min(limit, 100)
+    qs = Filament.objects.all()
+    if material:
+        qs = qs.filter(type__iexact=material)
+    if sub_type:
+        qs = qs.filter(sub_type__iexact=sub_type)
+    if brand:
+        qs = qs.filter(brand__icontains=brand)
+    if color:
+        qs = qs.filter(color__icontains=color)
+    if loaded_only:
+        qs = qs.filter(is_loaded_in_ams=True)
+    if min_remaining_percent is not None:
+        qs = qs.exclude(remaining_percent=-1).filter(remaining_percent__gte=min_remaining_percent)
+    if min_remaining_grams is not None:
+        qs = qs.exclude(remaining_percent=-1).exclude(remaining_weight_grams__isnull=True).filter(remaining_weight_grams__gte=min_remaining_grams)
+
+    spools = [_spool_summary(f, include_history) for f in qs.order_by("-remaining_percent")[:limit]]
+    return {"generated_at": timezone.now().isoformat(), "filaments": spools}
+
+
+def _spool_summary(f, include_history):
+    value, status = remaining_status(f.remaining_percent)
+    grams = f.remaining_weight_grams
+    entry = {
+        "id": f.id,
+        "material": f.type,
+        "sub_type": f.sub_type,
+        "brand": f.brand,
+        "color_name": f.color,
+        "color_hex": f.color_hex,
+        "is_transparent": f.is_transparent,
+        "remaining_percent": value,
+        "remaining_status": status,
+        "remaining_grams": grams if (grams is not None and grams >= 0) else None,
+        "ams_location": (
+            {"unit_id": f.ams_unit_id, "unit_type": f.ams_type, "slot": f.current_tray_id}
+            if f.is_loaded_in_ams else None
+        ),
+        "last_used": f.last_used.isoformat() if f.last_used else None,
+    }
+    if include_history:
+        entry["recent_usage"] = [
+            {
+                "job_id": u.print_job_id,
+                "job_name": u.print_job.project_name,
+                "started_at": u.print_job.start_time.isoformat(),
+                "consumed_grams": normalize_consumed(
+                    u.starting_percent, u.ending_percent, f.initial_weight_grams
+                )[1],
+            }
+            for u in f.print_usages.select_related("print_job").order_by("-print_job__start_time")[:3]
+        ]
+    return entry
+
+
+def get_filament(filament_id: int) -> dict:
+    """One spool in full, plus its complete print-usage history.
+
+    Usage rows with corrupt consumption (negative, or derived from the -1
+    "unread" sentinel) are returned with consumed_grams null and
+    data_quality "unreliable" rather than dropped -- the print still
+    happened, and hiding it would misstate the spool's history. Raises
+    Filament.DoesNotExist if filament_id is unknown.
+    """
+    f = Filament.objects.get(id=filament_id)
+    summary = _spool_summary(f, include_history=False)
+
+    usages = []
+    for u in f.print_usages.select_related("print_job").order_by("-print_job__start_time"):
+        consumed_percent, consumed_grams, quality = normalize_consumed(
+            u.starting_percent, u.ending_percent, f.initial_weight_grams
+        )
+        usages.append({
+            "job_id": u.print_job_id,
+            "job_name": u.print_job.project_name,
+            "started_at": u.print_job.start_time.isoformat(),
+            "consumed_percent": consumed_percent,
+            "consumed_grams": consumed_grams,
+            "data_quality": quality,
+        })
+
+    return {
+        **summary,
+        "generated_at": timezone.now().isoformat(),
+        "purchase_date": f.purchase_date.isoformat() if f.purchase_date else None,
+        "usage_history": usages,
+    }

bambu_run/mcp/jobs.py

@@ -0,0 +1,214 @@
+"""list_print_jobs, get_print_job, get_print_statistics -- the
+read_print_history capability tools."""
+from __future__ import annotations
+
+import statistics
+from datetime import timedelta
+
+from django.db.models import Avg, Count, ExpressionWrapper, F, FloatField, Q, Sum
+from django.utils import timezone
+
+from bambu_run.mcp.shapes import display_name, normalize_consumed, resolve_printer
+from bambu_run.models import BambuCloudTask, FilamentUsage, Filament, PrintJob
+
+
+def list_print_jobs(
+    status: str | None = None,
+    days: int = 30,
+    name_contains: str | None = None,
+    printer: str | int | None = None,
+    limit: int = 20,
+) -> dict:
+    """Print job history, most recent first.
+
+    status filters on FINISH/FAILED/CANCELLED, or "running" for the
+    in-flight job (final_status is null while printing). filament_grams
+    prefers Bambu Cloud's reported weight over the noisier AMS percentage
+    estimate -- check filament_grams_source ("cloud" vs None, meaning no
+    cloud data) before trusting precision. cost is always null until
+    filament purchase prices are recorded. limit has a hard ceiling of 100,
+    independent of days -- a busy printer's older jobs simply won't be
+    returned by one call; narrow days or increase limit deliberately.
+    """
+    limit = min(limit, 100)
+    now = timezone.now()
+    qs = PrintJob.objects.select_related("cloud_task", "device")
+    if printer is not None:
+        qs = qs.filter(device__in=resolve_printer(printer))
+    since = now - timedelta(days=days)
+    qs = qs.filter(start_time__gte=since)
+    if status == "running":
+        qs = qs.filter(final_status__isnull=True)
+    elif status:
+        qs = qs.filter(final_status=status)
+    if name_contains:
+        qs = qs.filter(project_name__icontains=name_contains)
+
+    jobs = list(qs.order_by("-start_time")[:limit])
+    return {"generated_at": now.isoformat(), "jobs": [_job_summary(j) for j in jobs]}
+
+
+def _job_summary(job):
+    cloud = job.cloud_task
+    est_min = int(cloud.cost_time_seconds / 60) if cloud and cloud.cost_time_seconds else None
+    grams = None
+    grams_source = None
+    if cloud and cloud.weight_grams is not None:
+        grams = float(cloud.weight_grams)
+        grams_source = "cloud"
+    return {
+        "id": job.id,
+        "display_name": display_name(
+            cloud.design_title if cloud else "", cloud.plate_title if cloud else "", job.project_name
+        ),
+        "status": job.final_status,
+        "printer": job.device.name,
+        "started_at": job.start_time.isoformat(),
+        "ended_at": job.end_time.isoformat() if job.end_time else None,
+        "duration_min": job.duration_minutes,
+        "estimated_duration_min": est_min,
+        "layers_total": job.total_layers,
+        "filament_grams": grams,
+        "filament_grams_source": grams_source,
+        "cost": None,
+        "has_cover_image": bool(cloud and cloud.cover_url),
+    }
+
+
+def get_print_job(job_id: int) -> dict:
+    """Full detail for one print job: the list_print_jobs fields plus
+    per-slot filament breakdown, estimate-vs-actual delta, and linked cloud
+    metadata. Raises PrintJob.DoesNotExist if job_id is unknown.
+    """
+    now = timezone.now()
+    job = PrintJob.objects.select_related("cloud_task", "device").get(id=job_id)
+    summary = _job_summary(job)
+
+    delta_percent = None
+    if summary["estimated_duration_min"] and summary["duration_min"]:
+        delta_percent = round(
+            100 * (summary["duration_min"] - summary["estimated_duration_min"])
+            / summary["estimated_duration_min"],
+            1,
+        )
+
+    usages = []
+    for u in job.filament_usages.select_related("filament"):
+        consumed_percent, consumed_grams, quality = normalize_consumed(
+            u.starting_percent, u.ending_percent,
+            u.filament.initial_weight_grams if u.filament else None,
+        )
+        usages.append({
+            "tray_id": u.tray_id,
+            "ams_unit_id": u.ams_unit_id,
+            "filament_id": u.filament_id,
+            "consumed_percent": consumed_percent,
+            "consumed_grams": consumed_grams,
+            "data_quality": quality,
+            "is_primary": u.is_primary,
+        })
+
+    cloud = job.cloud_task
+    return {
+        **summary,
+        "generated_at": now.isoformat(),
+        "estimate_delta_percent": delta_percent,
+        "filament_usages": usages,
+        "cloud_design_title": cloud.design_title if cloud else None,
+        "cover_url": cloud.cover_url if cloud and cloud.cover_url else None,
+    }
+
+
+def get_print_statistics(days: int = 30, printer: str | int | None = None) -> dict:
+    """Aggregate print analytics over a window: success rate, total time and
+    filament, estimate accuracy, failure patterns, and material mix.
+
+    cost_total is always null in this phase -- cost_available: false tells
+    the agent to explain the gap rather than report zero spend.
+    most_used_material answers "what did I print most" directly for the
+    requested window (days=7 for last week, days=30 for last month).
+    """
+    now = timezone.now()
+    since = now - timedelta(days=days)
+    qs = PrintJob.objects.filter(start_time__gte=since)
+    if printer is not None:
+        qs = qs.filter(device__in=resolve_printer(printer))
+
+    total_jobs = qs.count()
+    status_counts = {
+        (row["final_status"] or "running"): row["n"]
+        for row in qs.values("final_status").annotate(n=Count("id"))
+    }
+    finished = status_counts.get("FINISH", 0)
+    success_rate = round(finished / total_jobs, 3) if total_jobs else None
+
+    total_duration = qs.aggregate(s=Sum("duration_minutes"))["s"] or 0
+
+    cloud_qs = BambuCloudTask.objects.filter(print_jobs__in=qs).distinct()
+    total_grams = cloud_qs.aggregate(s=Sum("weight_grams"))["s"]
+    jobs_without_cloud = qs.filter(cloud_task__isnull=True).count()
+
+    # Same exclusion rules as normalize_consumed (starting_percent must be a
+    # real reading, and consumption must be non-negative), expressed as a DB
+    # filter -- a Python helper can't run inside an aggregate expression.
+    valid_usage = (
+        Q(starting_percent__gt=F("ending_percent"))
+        & ~Q(starting_percent=-1)
+        & ~Q(ending_percent=-1)
+        & Q(filament__initial_weight_grams__isnull=False)
+    )
+    material_rows = (
+        FilamentUsage.objects.filter(print_job__in=qs)
+        .filter(valid_usage)
+        .annotate(
+            grams=ExpressionWrapper(
+                (F("starting_percent") - F("ending_percent")) * F("filament__initial_weight_grams") / 100.0,
+                output_field=FloatField(),
+            )
+        )
+        .values("filament__type")
+        .annotate(total_grams=Sum("grams"))
+        .order_by("-total_grams")
+    )
+    material_mix = {row["filament__type"]: round(row["total_grams"], 2) for row in material_rows if row["filament__type"]}
+    most_used_material = next(iter(material_mix), None)
+
+    failed = qs.filter(final_status="FAILED")
+    failure_names = list(failed.values("project_name").annotate(n=Count("id")).order_by("-n")[:5])
+
+    ratio_qs = qs.filter(
+        duration_minutes__isnull=False, cloud_task__cost_time_seconds__gt=0
+    ).annotate(
+        ratio=ExpressionWrapper(
+            F("duration_minutes") / (F("cloud_task__cost_time_seconds") / 60.0),
+            output_field=FloatField(),
+        )
+    )
+    mean_ratio = ratio_qs.aggregate(avg=Avg("ratio"))["avg"]
+    # median has no portable DB-level equivalent across backends; this list
+    # is already bounded by the same days/printer filters as everything else.
+    ratios = list(ratio_qs.values_list("ratio", flat=True))
+    median_ratio = statistics.median(ratios) if ratios else None
+
+    return {
+        "generated_at": now.isoformat(),
+        "window_days": days,
+        "total_jobs": total_jobs,
+        "success_rate": success_rate,
+        "status_counts": status_counts,
+        "total_print_minutes": total_duration,
+        "total_filament_grams": float(total_grams) if total_grams is not None else None,
+        "jobs_without_cloud_data": jobs_without_cloud,
+        "estimate_accuracy": {
+            "mean_ratio": round(mean_ratio, 2) if mean_ratio is not None else None,
+            "median_ratio": round(median_ratio, 2) if median_ratio is not None else None,
+        },
+        "failures": {
+            "count": failed.count(),
+            "top_failing_projects": [{"name": f["project_name"], "count": f["n"]} for f in failure_names],
+        },
+        "material_mix_grams": material_mix,
+        "most_used_material": most_used_material,
+        "cost_total": None,
+        "cost_available": False,
+    }

bambu_run/mcp/shapes.py

@@ -0,0 +1,146 @@
+"""Shared output-shaping helpers enforcing the spec's data-correction rules.
+
+Every rule here traces to a verified finding in "Bambu-Run MCP Tool Surface"
+(the "Data Reality Findings" section) -- not a general defensive-programming
+habit. Do not relax these without re-checking against real data.
+"""
+from __future__ import annotations
+
+import statistics
+
+STALE_THRESHOLD_SECONDS = 120
+LOW_REMAINING_THRESHOLD = 20
+
+
+def freshness(timestamp, now):
+    """(data_age_seconds, stale) for a reading timestamp relative to `now`.
+
+    `now` is passed in rather than computed here so a caller building a
+    multi-field response reuses one `timezone.now()` call throughout,
+    keeping `generated_at` and every `data_age_seconds` mutually consistent.
+    """
+    age_seconds = (now - timestamp).total_seconds()
+    return {
+        "data_age_seconds": int(age_seconds),
+        "stale": age_seconds > STALE_THRESHOLD_SECONDS,
+    }
+
+
+def remaining_status(percent):
+    """Classify one raw remaining-percent reading, honoring the -1 sentinel.
+
+    Returns (value, status): value is percent-or-None; status is one of
+    "unread" | "empty" | "low" | "ok". -1 and None both mean "no RFID
+    reading yet" -- the spec rules out -1 meaning "low" (jobs 62/63/70/77
+    show start=-1% followed by end=90-100%, which a genuinely low reading
+    could not do).
+    """
+    if percent is None or percent == -1:
+        return None, "unread"
+    percent = int(round(percent))
+    if percent <= 0:
+        return 0, "empty"
+    if percent <= LOW_REMAINING_THRESHOLD:
+        return percent, "low"
+    return percent, "ok"
+
+
+def smoothed_remaining_percent(readings):
+    """Median of the 5 most recent remain_percent readings for one tray.
+
+    `readings` is most-recent-first. -1 sentinel readings are excluded
+    before the median, not treated as zero (spec: AMS remain_percent swings
+    22-35% mid-print on real RFID noise; median-of-5 rejects single-sample
+    spikes while still tracking genuine change).
+    """
+    recent = readings[:5]
+    valid = [r for r in recent if r is not None and r != -1]
+    if not valid:
+        return remaining_status(-1 if recent else None)
+    return remaining_status(statistics.median(valid))
+
+
+def classify_alerts(code_counts, total_rows):
+    """Turn {(attr, code): occurrences} window counts into the alerts array.
+
+    Frequency heuristic (spec: HMS codes are stateful, not event-based --
+    the two most common codes alone accounted for over 80% of all alert
+    occurrences in the verified data, in long unbroken runs, not isolated
+    events): a code present in >20% of window rows is "persistent" (a
+    likely standing advisory, surfaced once with a count); <5% is
+    "transient" (possibly actionable). The 5-20% band is omitted rather
+    than guessed at. severity is always "unknown" -- nothing in Bambu-Run
+    decodes these codes yet; do not assert "fine" or "faulty" from them.
+    """
+    alerts = []
+    for (attr, code), count in code_counts.items():
+        ratio = count / total_rows if total_rows else 0
+        if ratio > 0.20:
+            kind = "persistent"
+        elif ratio < 0.05:
+            kind = "transient"
+        else:
+            continue
+        alerts.append({
+            "attr": attr,
+            "code": code,
+            "kind": kind,
+            "occurrences": count,
+            "severity": "unknown",
+        })
+    return alerts
+
+
+def display_name(design_title, plate_title, project_name):
+    """design_title -> plate_title -> project_name fallback.
+
+    Spec finding: design_title is empty on 4 of 5 recent BambuCloudTask rows.
+    """
+    return design_title or plate_title or project_name
+
+
+def normalize_consumed(starting_percent, ending_percent, initial_weight_grams):
+    """FilamentUsage consumption, honest about the -1 sentinel and RFID jitter.
+
+    Returns (consumed_percent, consumed_grams, data_quality). data_quality is
+    "ok" or "unreliable". A negative or sentinel-derived result becomes
+    (None, None, "unreliable") -- never a negative figure, never a
+    fabricated zero (spec: 24/269 usage rows go negative; a -1 sentinel
+    subtracted as a real value produces e.g. consumed=-101%).
+    """
+    if (
+        starting_percent is None or starting_percent == -1
+        or ending_percent is None or ending_percent == -1
+    ):
+        return None, None, "unreliable"
+    consumed_percent = starting_percent - ending_percent
+    if consumed_percent < 0:
+        return None, None, "unreliable"
+    consumed_grams = None
+    if initial_weight_grams:
+        consumed_grams = int(initial_weight_grams * (consumed_percent / 100.0))
+    return consumed_percent, consumed_grams, "ok"
+
+
+def resolve_printer(printer):
+    """Resolve an id, exact name, or None to a Printer queryset.
+
+    None returns all active printers. Raises Printer.DoesNotExist if a
+    specific id/name was given and nothing matched -- MCP tool wrappers let
+    this propagate as a clear tool error rather than returning empty data
+    that looks like "no printers exist".
+    """
+    from bambu_run.models import Printer
+
+    if printer is None:
+        return Printer.objects.filter(is_active=True)
+
+    if isinstance(printer, int) or (isinstance(printer, str) and printer.isdigit()):
+        by_id = Printer.objects.filter(pk=int(printer))
+        if by_id.exists():
+            return by_id
+
+    by_name = Printer.objects.filter(name=printer)
+    if not by_name.exists():
+        raise Printer.DoesNotExist(f"No printer matching {printer!r}")
+    return by_name

bambu_run/mcp/status.py

@@ -0,0 +1,268 @@
+"""get_printer_status, get_metrics_history, get_hardware_health -- the
+read_printer capability tools."""
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import timedelta
+
+from django.db.models import Avg
+from django.utils import timezone
+
+from bambu_run.mcp.shapes import classify_alerts, freshness, resolve_printer, smoothed_remaining_percent
+from bambu_run.models import Hotend, PrinterMetrics
+from bambu_run.utils import strip_color_padding
+
+ALERT_WINDOW_HOURS = 24
+
+
+def get_printer_status(printer: str | int | None = None) -> dict:
+    """Current live status of one or all printers: temps, print progress, AMS slots, and alerts.
+
+    Pass a printer id or exact name to scope to one printer; omit to get every
+    active printer. Numeric percentages under ams_units[].slots[] are smoothed
+    (median of the 5 most recent readings) and paired with a remain_status
+    field ("ok"/"low"/"empty"/"unread") -- treat "unread" as "no RFID reading
+    yet", not as "empty". alerts are frequency-classified, not raw fault
+    codes; severity is always "unknown" until Bambu's HMS codes are mapped,
+    so never assert the printer is fine or faulty purely from an alert entry.
+    """
+    now = timezone.now()
+    printers = resolve_printer(printer)
+
+    entries = []
+    for p in printers:
+        metric = PrinterMetrics.objects.filter(device=p).order_by("-timestamp").first()
+        if metric is None:
+            entries.append({
+                "id": p.id, "name": p.name, "model": p.model, "location": p.location,
+                "data_age_seconds": None, "stale": True, "state": "no_data",
+                "job": None, "temperatures": None, "ams_units": [], "alerts": [],
+                "fans": None, "chamber_light": None,
+            })
+            continue
+
+        entries.append({
+            "id": p.id,
+            "name": p.name,
+            "model": p.model,
+            "location": p.location,
+            **freshness(metric.timestamp, now),
+            "state": metric.gcode_state or "UNKNOWN",
+            "job": _job_info(metric, now),
+            "temperatures": _temperatures(metric),
+            "ams_units": _ams_units(p, metric),
+            "alerts": _alerts_for_printer(p, now),
+            "fans": {
+                "cooling_percent": metric.cooling_fan_speed,
+                "aux_percent": metric.big_fan1_speed,
+                "chamber_percent": metric.big_fan2_speed,
+            },
+            "chamber_light": metric.chamber_light,
+        })
+
+    return {"generated_at": now.isoformat(), "printers": entries}
+
+
+def _job_info(metric, now):
+    if metric.gcode_state != "RUNNING":
+        return None
+    remaining = metric.remaining_time_min
+    finish_at = (now + timedelta(minutes=remaining)).isoformat() if remaining is not None else None
+    return {
+        "name": metric.subtask_name,
+        "print_percent": metric.print_percent,
+        "layer_current": metric.layer_num,
+        "layer_total": metric.total_layer_num,
+        "remaining_time_min": remaining,
+        "estimated_finish_at": finish_at,
+    }
+
+
+def _temperatures(metric):
+    def f(v):
+        return float(v) if v is not None else None
+    return {
+        "nozzle_c": f(metric.nozzle_temp), "nozzle_target_c": f(metric.nozzle_target_temp),
+        "nozzle_left_c": f(metric.nozzle_temp_left), "nozzle_left_target_c": f(metric.nozzle_target_temp_left),
+        "bed_c": f(metric.bed_temp), "bed_target_c": f(metric.bed_target_temp),
+        "chamber_c": f(metric.chamber_temp),
+    }
+
+
+def _safe_tray_id(value, default=-1):
+    """Coerce a tray_id from the filaments JSON to int, defaulting on anything
+    that isn't a clean int (missing, None, or malformed) rather than crashing
+    the whole tool from one bad row."""
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _ams_units(printer, latest_metric):
+    """Build ams_units from the latest metric's filaments, with remain_percent
+    smoothed against the printer's 5 most recent metric rows.
+
+    Per-unit humidity/temp come from PrinterMetrics.ams_units (a JSONField with
+    one entry per physical AMS unit) -- NOT from PrinterMetrics.ams_humidity_raw/
+    .ams_temp, which the collector only ever populates from unit 0's reading and
+    would otherwise stamp one unit's humidity onto every unit.
+    """
+    recent_filament_lists = list(
+        PrinterMetrics.objects.filter(device=printer)
+        .order_by("-timestamp")
+        .values_list("filaments", flat=True)[:5]
+    )
+    readings = defaultdict(list)
+    for filaments in recent_filament_lists:
+        for f in filaments or []:
+            key = (f.get("ams_unit_id"), _safe_tray_id(f.get("tray_id")))
+            readings[key].append(f.get("remain_percent"))
+
+    unit_readings = {
+        u.get("unit_id"): u for u in (latest_metric.ams_units or [])
+    }
+
+    units = {}
+    for f in latest_metric.filaments or []:
+        unit_id = f.get("ams_unit_id")
+        tray_id = _safe_tray_id(f.get("tray_id"))
+        value, status = smoothed_remaining_percent(readings[(unit_id, tray_id)])
+        unit_reading = unit_readings.get(unit_id, {})
+        unit = units.setdefault(unit_id, {
+            "unit_id": unit_id,
+            "type": f.get("ams_type", ""),
+            "humidity_raw": unit_reading.get("humidity_raw"),
+            "temp_c": float(unit_reading["temp"]) if unit_reading.get("temp") is not None else None,
+            "slots": [],
+        })
+        color = f.get("color")
+        unit["slots"].append({
+            "slot": tray_id,
+            "label": f.get("slot", ""),
+            "material": f.get("type", ""),
+            "sub_type": f.get("sub_type", ""),
+            "color_hex": f"#{strip_color_padding(color)}" if color else None,
+            "remain_percent": value,
+            "remain_status": status,
+        })
+    return list(units.values())
+
+
+def _alerts_for_printer(printer, now, window_hours=ALERT_WINDOW_HOURS):
+    since = now - timedelta(hours=window_hours)
+    window = PrinterMetrics.objects.filter(device=printer, timestamp__gte=since)
+    total = window.count()
+    counts = defaultdict(int)
+    for hms_list in window.exclude(hms=[]).values_list("hms", flat=True):
+        for entry in hms_list:
+            counts[(entry.get("attr"), entry.get("code"))] += 1
+    return classify_alerts(counts, total)
+
+
+def get_metrics_history(
+    printer: str | int | None = None,
+    hours: int = 6,
+    metrics: list[str] | None = None,
+    max_points: int = 300,
+) -> dict:
+    """Sampled time-series history for one printer: temps, humidity, progress, fans.
+
+    metrics selects which series to include: any of "temps", "humidity",
+    "progress", "fans" (omit for all four). Server-side sampled to at most
+    max_points (hard ceiling 500) evenly across the window -- do not expect
+    one point per collector poll on a long window. printer defaults to the
+    first active printer when omitted; pass an id or name explicitly if more
+    than one printer is registered.
+    """
+    from bambu_run.views import sample_metrics  # reuse: same thinning the dashboard already uses
+
+    now = timezone.now()
+    max_points = min(max_points, 500)
+    target = resolve_printer(printer).first()
+    if target is None:
+        return {"generated_at": now.isoformat(), "printer": None, "timestamps": [], "series": {}}
+
+    since = now - timedelta(hours=hours)
+    rows = list(
+        PrinterMetrics.objects.filter(device=target, timestamp__gte=since).order_by("timestamp")
+    )
+    rows = sample_metrics(rows, max_points=max_points)
+
+    wanted = set(metrics) if metrics else {"temps", "humidity", "progress", "fans"}
+    series = {}
+    if "temps" in wanted:
+        series["nozzle_c"] = [float(r.nozzle_temp) if r.nozzle_temp is not None else None for r in rows]
+        series["bed_c"] = [float(r.bed_temp) if r.bed_temp is not None else None for r in rows]
+        series["chamber_c"] = [float(r.chamber_temp) if r.chamber_temp is not None else None for r in rows]
+    if "humidity" in wanted:
+        series["ams_humidity_raw"] = [r.ams_humidity_raw for r in rows]
+    if "progress" in wanted:
+        series["print_percent"] = [r.print_percent for r in rows]
+    if "fans" in wanted:
+        series["cooling_fan_percent"] = [r.cooling_fan_speed for r in rows]
+
+    return {
+        "generated_at": now.isoformat(),
+        "printer": {"id": target.id, "name": target.name},
+        "timestamps": [r.timestamp.isoformat() for r in rows],
+        "series": series,
+    }
+
+
+def get_hardware_health(printer: str | int | None = None, days: int = 7) -> dict:
+    """Maintenance-relevant observations for one printer: wifi, AMS humidity,
+    alert summary, hotend inventory, and collector coverage.
+
+    Returns observations only -- it does not render a verdict; the calling
+    agent decides what the numbers mean. wear_percent is deliberately absent
+    from each hotend entry: verified constant at 100.00 across every
+    hotend's entire history, so it carries no information.
+    """
+    now = timezone.now()
+    target = resolve_printer(printer).first()
+    if target is None:
+        return {"printer": None}
+
+    since = now - timedelta(days=days)
+    window = PrinterMetrics.objects.filter(device=target, timestamp__gte=since)
+    total = window.count()
+    latest = window.order_by("-timestamp").first()
+
+    wifi_avg = window.exclude(wifi_signal_dbm__isnull=True).aggregate(avg=Avg("wifi_signal_dbm"))["avg"]
+    humidity_avg = window.exclude(ams_humidity_raw__isnull=True).aggregate(avg=Avg("ams_humidity_raw"))["avg"]
+
+    counts = defaultdict(int)
+    for hms_list in window.exclude(hms=[]).values_list("hms", flat=True):
+        for entry in hms_list:
+            counts[(entry.get("attr"), entry.get("code"))] += 1
+
+    hotends = [
+        {
+            "id": h.id,
+            "serial_number": h.serial_number,
+            "position": "toolhead" if h.is_toolhead else (f"slot_{h.slot_number}" if h.slot_number else "rack"),
+            "used_time_hours": round(h.used_time_seconds / 3600, 1),
+            "last_filament_profile_id": h.last_filament_profile_id or None,
+        }
+        for h in Hotend.objects.filter(printer=target)
+    ]
+
+    expected_polls = int(days * 24 * 60 * 60 / 30)  # 30s measured collector cadence
+
+    return {
+        "printer": {"id": target.id, "name": target.name},
+        "generated_at": now.isoformat(),
+        "wifi_signal_dbm": {
+            "current": latest.wifi_signal_dbm if latest else None,
+            "window_avg": round(wifi_avg, 1) if wifi_avg is not None else None,
+        },
+        "ams_humidity_raw_window_avg": round(humidity_avg, 1) if humidity_avg is not None else None,
+        "alerts": classify_alerts(counts, total),
+        "hotends": hotends,
+        "collector": {
+            "polls_in_window": total,
+            "expected_polls": expected_polls,
+            "coverage_ratio": round(total / expected_polls, 2) if expected_polls else None,
+        },
+    }

bambu_run/mcp/toolkit.py

@@ -0,0 +1,28 @@
+"""Assembles the 8-tool bambu_run MCP toolkit, capability-tagged per the spec."""
+from __future__ import annotations
+
+from mcp_gate import Toolkit
+
+from bambu_run.mcp.filaments import get_filament, list_filaments
+from bambu_run.mcp.jobs import get_print_job, get_print_statistics, list_print_jobs
+from bambu_run.mcp.status import get_hardware_health, get_metrics_history, get_printer_status
+
+
+def build_toolkit() -> Toolkit:
+    """Build the bambu_run MCP toolkit: 8 tools across read_printer,
+    read_print_history, and read_filament.
+
+    Called fresh by each entry point rather than shared as a module-level
+    singleton, so tests (and multiple concurrent Gates, if ever needed) get
+    an isolated Toolkit instance rather than sharing registration state.
+    """
+    kit = Toolkit("bambu_run")
+    kit.tool(requires="read_printer")(get_printer_status)
+    kit.tool(requires="read_printer")(get_metrics_history)
+    kit.tool(requires="read_printer")(get_hardware_health)
+    kit.tool(requires="read_print_history")(list_print_jobs)
+    kit.tool(requires="read_print_history")(get_print_job)
+    kit.tool(requires="read_print_history")(get_print_statistics)
+    kit.tool(requires="read_filament")(list_filaments)
+    kit.tool(requires="read_filament")(get_filament)
+    return kit

bambu_run/mcp_tools.py

@@ -1,728 +0,0 @@
-"""
-Pure Django ORM query functions for MCP tools.
-
-Zero dependency on the `mcp` package — returns markdown strings.
-RAE can reuse these directly.
-"""
-
-from datetime import timedelta
-from decimal import Decimal
-from zoneinfo import ZoneInfo
-
-from django.db.models import Avg, Count, Max, Min, Q, Sum
-from django.utils import timezone
-
-from .conf import app_settings
-
-
-def _local_dt(dt, fmt="%Y-%m-%d %H:%M %Z"):
-    """Convert a UTC-aware datetime to the configured local timezone for display."""
-    if dt is None:
-        return "—"
-    tz = ZoneInfo(app_settings.TIMEZONE)
-    return dt.astimezone(tz).strftime(fmt)
-
-
-def _redact(value, label="[redacted]"):
-    """Redact sensitive values if MCP_HIDE_SENSITIVE is enabled."""
-    if app_settings.MCP_HIDE_SENSITIVE:
-        return label
-    return value
-
-
-def _job_name(job):
-    """Return the best available display name for a print job.
-
-    Prefers cloud design_title (e.g., 'Planetary Gears Finger Fidget Spinners')
-    over the MQTT subtask_name (e.g., 'All variants at 0.16mm high quality').
-    Falls back to project_name for local/SD prints with no cloud task.
-    """
-    if job.cloud_task_id and job.cloud_task and job.cloud_task.design_title:
-        return job.cloud_task.design_title
-    return job.project_name
-
-
-def _format_duration(minutes):
-    """Format minutes into human-readable duration."""
-    if minutes is None:
-        return "Unknown"
-    hours, mins = divmod(int(minutes), 60)
-    if hours > 0:
-        return f"{hours}h {mins}m"
-    return f"{mins}m"
-
-
-def _format_temp(temp):
-    """Format temperature value."""
-    if temp is None:
-        return "N/A"
-    return f"{temp}°C"
-
-
-# ─── Tools ───────────────────────────────────────────────────────────────────
-
-
-def get_printer_status(printer_id=None):
-    """Current live status of printer(s) including temps, progress, AMS, errors."""
-    from .models import Printer, PrinterMetrics
-
-    printers = Printer.objects.filter(is_active=True)
-    if printer_id:
-        printers = printers.filter(id=printer_id)
-
-    if not printers.exists():
-        return "No printers found."
-
-    parts = []
-    for printer in printers:
-        metric = PrinterMetrics.objects.filter(device=printer).first()
-        if not metric:
-            parts.append(f"## {printer.name}\n**No data available yet.**\n")
-            continue
-
-        state = metric.gcode_state or "Unknown"
-        lines = [f"## Printer Status: {printer.name}"]
-        lines.append(f"**Model**: {printer.model} | **Serial**: {_redact(printer.serial_number)}")
-        lines.append(f"**IP**: {_redact(printer.ip_address)} | **Location**: {printer.location or 'N/A'}")
-        lines.append(f"**State**: {state}")
-
-        if metric.print_percent is not None and state == "RUNNING":
-            layer_info = ""
-            if metric.layer_num is not None and metric.total_layer_num:
-                layer_info = f" (Layer {metric.layer_num}/{metric.total_layer_num})"
-            lines.append(f"**Progress**: {metric.print_percent}%{layer_info}")
-            if metric.subtask_name:
-                lines.append(f"**Project**: {metric.subtask_name}")
-            if metric.remaining_time_min:
-                lines.append(f"**ETA**: {_format_duration(metric.remaining_time_min)} remaining")
-
-        # Temperatures
-        lines.append("")
-        lines.append("### Temperatures")
-        lines.append("| Component | Current | Target |")
-        lines.append("|-----------|---------|--------|")
-        lines.append(f"| Nozzle | {_format_temp(metric.nozzle_temp)} | {_format_temp(metric.nozzle_target_temp)} |")
-        lines.append(f"| Bed | {_format_temp(metric.bed_temp)} | {_format_temp(metric.bed_target_temp)} |")
-        lines.append(f"| Chamber | {_format_temp(metric.chamber_temp)} | - |")
-
-        # AMS filaments from JSON
-        if metric.filaments:
-            lines.append("")
-            lines.append("### AMS Slots")
-            lines.append("| Slot | Material | Color | Remaining |")
-            lines.append("|------|----------|-------|-----------|")
-            for f in metric.filaments:
-                slot = f.get("slot", "?")
-                ftype = f.get("sub_type") or f.get("type", "?")
-                color = f.get("color", "")
-                color_display = f"#{color[:6]}" if color and len(color) >= 6 else "?"
-                remain = f.get("remain_percent", "?")
-                lines.append(f"| {slot} | {ftype} | {color_display} | {remain}% |")
-
-        # Errors
-        if metric.has_errors or metric.hms:
-            lines.append("")
-            lines.append("### Alerts")
-            if metric.print_error:
-                lines.append(f"- Print error code: {metric.print_error}")
-            if metric.hms:
-                for msg in metric.hms[:5]:
-                    lines.append(f"- HMS: {msg}")
-
-        lines.append(f"\n*Last updated: {_local_dt(metric.timestamp, '%Y-%m-%d %H:%M:%S %Z')}*")
-        parts.append("\n".join(lines))
-
-    return "\n\n---\n\n".join(parts)
-
-
-def list_printers():
-    """List all registered printers."""
-    from .models import Printer
-
-    printers = Printer.objects.all()
-    if not printers.exists():
-        return "No printers registered."
-
-    lines = ["# Printers", ""]
-    lines.append("| ID | Name | Model | Active | Serial | IP | Location |")
-    lines.append("|----|------|-------|--------|--------|----|----------|")
-    for p in printers:
-        lines.append(
-            f"| {p.id} | {p.name} | {p.model} | "
-            f"{'Yes' if p.is_active else 'No'} | "
-            f"{_redact(p.serial_number)} | {_redact(p.ip_address)} | "
-            f"{p.location or '-'} |"
-        )
-    return "\n".join(lines)
-
-
-def get_print_history(status=None, days=None, project_name=None, limit=20):
-    """Print job history with optional filters."""
-    from .models import PrintJob
-
-    qs = PrintJob.objects.select_related("device", "cloud_task")
-
-    if status:
-        qs = qs.filter(final_status__iexact=status)
-    if days:
-        cutoff = timezone.now() - timedelta(days=int(days))
-        qs = qs.filter(start_time__gte=cutoff)
-    if project_name:
-        qs = qs.filter(
-            Q(project_name__icontains=project_name)
-            | Q(cloud_task__design_title__icontains=project_name)
-        )
-
-    jobs = qs[:int(limit)]
-    if not jobs:
-        return "No print jobs found matching the criteria."
-
-    lines = ["# Print History", ""]
-    lines.append("| ID | Project | Printer | Status | Progress | Duration | Started |")
-    lines.append("|----|---------|---------|--------|----------|----------|---------|")
-    for j in jobs:
-        lines.append(
-            f"| {j.id} | {_job_name(j)} | {j.device.name} | "
-            f"{j.final_status or 'In Progress'} | {j.completion_percent}% | "
-            f"{_format_duration(j.duration_minutes)} | "
-            f"{_local_dt(j.start_time, '%Y-%m-%d %H:%M')} |"
-        )
-    return "\n".join(lines)
-
-
-def get_print_job_detail(job_id):
-    """Single job detail including filament usage."""
-    from .models import FilamentUsage, PrintJob
-
-    try:
-        job = PrintJob.objects.select_related("device", "cloud_task").get(id=job_id)
-    except PrintJob.DoesNotExist:
-        return f"Print job #{job_id} not found."
-
-    lines = [f"# Print Job: {_job_name(job)}", ""]
-    if job.cloud_task and job.cloud_task.design_title and job.cloud_task.design_title != job.project_name:
-        lines.append(f"**Plate**: {job.project_name}")
-    lines.append(f"**Printer**: {job.device.name}")
-    lines.append(f"**Status**: {job.final_status or 'In Progress'}")
-    lines.append(f"**Progress**: {job.completion_percent}%")
-    if job.gcode_file:
-        lines.append(f"**G-code**: {job.gcode_file}")
-    lines.append(f"**Started**: {_local_dt(job.start_time, '%Y-%m-%d %H:%M:%S %Z')}")
-    if job.end_time:
-        lines.append(f"**Ended**: {_local_dt(job.end_time, '%Y-%m-%d %H:%M:%S %Z')}")
-    lines.append(f"**Duration**: {_format_duration(job.duration_minutes)}")
-    if job.total_layers:
-        lines.append(f"**Total Layers**: {job.total_layers}")
-
-    # Filament usage
-    usages = FilamentUsage.objects.select_related("filament").filter(print_job=job)
-    if usages.exists():
-        lines.append("")
-        lines.append("### Filament Usage")
-        lines.append("| Spool | Material | Color | Consumed | Grams |")
-        lines.append("|-------|----------|-------|----------|-------|")
-        for u in usages:
-            f = u.filament
-            lines.append(
-                f"| {f.brand} {f.type} | {f.sub_type or f.type} | "
-                f"{f.color} | {u.consumed_percent or 0}% | "
-                f"{u.consumed_grams or '-'}g |"
-            )
-
-    return "\n".join(lines)
-
-
-def list_filaments(type=None, brand=None, color=None, loaded_in_ams=None, low_filament=None):
-    """Filament inventory with optional filters."""
-    from .models import Filament
-
-    qs = Filament.objects.all()
-    if type:
-        qs = qs.filter(type__iexact=type)
-    if brand:
-        qs = qs.filter(brand__icontains=brand)
-    if color:
-        qs = qs.filter(color__icontains=color)
-    if loaded_in_ams is not None:
-        qs = qs.filter(is_loaded_in_ams=loaded_in_ams)
-    if low_filament:
-        qs = qs.filter(remaining_percent__lte=20)
-
-    filaments = qs[:50]
-    if not filaments:
-        return "No filaments found matching the criteria."
-
-    lines = ["# Filament Inventory", ""]
-    lines.append(f"*{qs.count()} spools total*\n")
-    lines.append("| ID | Brand | Type | Color | Remaining | In AMS | Last Used |")
-    lines.append("|----|-------|------|-------|-----------|--------|-----------|")
-    for f in filaments:
-        color_display = f"{f.color}"
-        if f.color_hex:
-            color_display += f" ({f.color_hex})"
-        last_used = _local_dt(f.last_used, "%Y-%m-%d") if f.last_used else "-"
-        lines.append(
-            f"| {f.id} | {f.brand} | {f.sub_type or f.type} | "
-            f"{color_display} | {f.remaining_percent}% | "
-            f"{'Yes' if f.is_loaded_in_ams else 'No'} | {last_used} |"
-        )
-    return "\n".join(lines)
-
-
-def get_filament_detail(filament_id):
-    """Single spool detail with usage history."""
-    from .models import Filament, FilamentUsage
-
-    try:
-        f = Filament.objects.get(id=filament_id)
-    except Filament.DoesNotExist:
-        return f"Filament #{filament_id} not found."
-
-    lines = [f"# Filament: {f.brand} {f.type} - {f.color}", ""]
-    lines.append(f"**Type**: {f.sub_type or f.type}")
-    lines.append(f"**Brand**: {f.brand}")
-    lines.append(f"**Color**: {f.color} ({f.color_hex or 'N/A'})")
-    lines.append(f"**Remaining**: {f.remaining_percent}%")
-    if f.remaining_weight_grams:
-        lines.append(f"**Remaining Weight**: {f.remaining_weight_grams}g / {f.initial_weight_grams or '?'}g")
-    lines.append(f"**In AMS**: {'Yes (slot ' + str(f.current_tray_id) + ')' if f.is_loaded_in_ams else 'No'}")
-    lines.append(f"**Created By**: {f.created_by}")
-    if f.tray_uuid:
-        lines.append(f"**Serial**: {_redact(f.tray_uuid)}")
-    if f.purchase_date:
-        lines.append(f"**Purchased**: {f.purchase_date}")
-    if f.notes:
-        lines.append(f"**Notes**: {f.notes}")
-
-    # Usage history
-    usages = FilamentUsage.objects.select_related("print_job").filter(filament=f).order_by("-print_job__start_time")[:10]
-    if usages.exists():
-        lines.append("")
-        lines.append("### Recent Print Usage")
-        lines.append("| Job | Date | Consumed | Grams |")
-        lines.append("|-----|------|----------|-------|")
-        for u in usages:
-            lines.append(
-                f"| {u.print_job.project_name} | "
-                f"{_local_dt(u.print_job.start_time, '%Y-%m-%d')} | "
-                f"{u.consumed_percent or 0}% | {u.consumed_grams or '-'}g |"
-            )
-
-    return "\n".join(lines)
-
-
-def get_temperature_history(printer_id=None, hours=6, metric="all"):
-    """Temperature trends as summary stats (avg/min/max) over recent hours."""
-    from .models import Printer, PrinterMetrics
-
-    cutoff = timezone.now() - timedelta(hours=int(hours))
-
-    qs = PrinterMetrics.objects.filter(timestamp__gte=cutoff)
-    if printer_id:
-        qs = qs.filter(device_id=printer_id)
-
-    if not qs.exists():
-        return f"No temperature data in the last {hours} hours."
-
-    printers = Printer.objects.filter(
-        id__in=qs.values_list("device_id", flat=True).distinct()
-    )
-
-    parts = [f"# Temperature History (last {hours}h)", ""]
-    for printer in printers:
-        pqs = qs.filter(device=printer)
-        stats = pqs.aggregate(
-            nozzle_avg=Avg("nozzle_temp"),
-            nozzle_min=Min("nozzle_temp"),
-            nozzle_max=Max("nozzle_temp"),
-            bed_avg=Avg("bed_temp"),
-            bed_min=Min("bed_temp"),
-            bed_max=Max("bed_temp"),
-            chamber_avg=Avg("chamber_temp"),
-            chamber_min=Min("chamber_temp"),
-            chamber_max=Max("chamber_temp"),
-        )
-
-        parts.append(f"## {printer.name}")
-        parts.append(f"*{pqs.count()} data points*\n")
-        parts.append("| Sensor | Avg | Min | Max |")
-        parts.append("|--------|-----|-----|-----|")
-
-        if metric in ("all", "nozzle"):
-            parts.append(
-                f"| Nozzle | {_format_temp(stats['nozzle_avg'])} | "
-                f"{_format_temp(stats['nozzle_min'])} | {_format_temp(stats['nozzle_max'])} |"
-            )
-        if metric in ("all", "bed"):
-            parts.append(
-                f"| Bed | {_format_temp(stats['bed_avg'])} | "
-                f"{_format_temp(stats['bed_min'])} | {_format_temp(stats['bed_max'])} |"
-            )
-        if metric in ("all", "chamber"):
-            parts.append(
-                f"| Chamber | {_format_temp(stats['chamber_avg'])} | "
-                f"{_format_temp(stats['chamber_min'])} | {_format_temp(stats['chamber_max'])} |"
-            )
-        parts.append("")
-
-    return "\n".join(parts)
-
-
-def get_filament_usage_stats(days=30, group_by="type"):
-    """Aggregate filament consumption statistics."""
-    from .models import FilamentUsage
-
-    cutoff = timezone.now() - timedelta(days=int(days))
-    qs = FilamentUsage.objects.filter(
-        print_job__start_time__gte=cutoff,
-        consumed_grams__isnull=False,
-    ).select_related("filament")
-
-    if not qs.exists():
-        return f"No filament usage data in the last {days} days."
-
-    lines = [f"# Filament Usage Stats (last {days} days)", ""]
-
-    if group_by == "type":
-        stats = (
-            qs.values("filament__type")
-            .annotate(
-                total_grams=Sum("consumed_grams"),
-                total_percent=Sum("consumed_percent"),
-                job_count=Count("print_job", distinct=True),
-            )
-            .order_by("-total_grams")
-        )
-        lines.append("| Type | Total Grams | Jobs | Avg Grams/Job |")
-        lines.append("|------|-------------|------|---------------|")
-        for s in stats:
-            avg = s["total_grams"] / s["job_count"] if s["job_count"] else 0
-            lines.append(
-                f"| {s['filament__type']} | {s['total_grams']}g | "
-                f"{s['job_count']} | {avg:.0f}g |"
-            )
-    elif group_by == "color":
-        stats = (
-            qs.values("filament__color", "filament__type")
-            .annotate(total_grams=Sum("consumed_grams"), job_count=Count("print_job", distinct=True))
-            .order_by("-total_grams")
-        )
-        lines.append("| Color | Type | Total Grams | Jobs |")
-        lines.append("|-------|------|-------------|------|")
-        for s in stats:
-            lines.append(
-                f"| {s['filament__color']} | {s['filament__type']} | "
-                f"{s['total_grams']}g | {s['job_count']} |"
-            )
-    elif group_by == "spool":
-        stats = (
-            qs.values("filament__id", "filament__brand", "filament__type", "filament__color")
-            .annotate(total_grams=Sum("consumed_grams"), job_count=Count("print_job", distinct=True))
-            .order_by("-total_grams")[:20]
-        )
-        lines.append("| Spool | Total Grams | Jobs |")
-        lines.append("|-------|-------------|------|")
-        for s in stats:
-            lines.append(
-                f"| {s['filament__brand']} {s['filament__type']} {s['filament__color']} | "
-                f"{s['total_grams']}g | {s['job_count']} |"
-            )
-
-    return "\n".join(lines)
-
-
-def get_printer_health(printer_id=None):
-    """Diagnostics: errors, humidity, wifi, recent failed prints."""
-    from .models import Printer, PrinterMetrics, PrintJob
-
-    printers = Printer.objects.filter(is_active=True)
-    if printer_id:
-        printers = printers.filter(id=printer_id)
-
-    if not printers.exists():
-        return "No printers found."
-
-    parts = ["# Printer Health Report", ""]
-    for printer in printers:
-        latest = PrinterMetrics.objects.filter(device=printer).first()
-        if not latest:
-            parts.append(f"## {printer.name}\n**No data available.**\n")
-            continue
-
-        parts.append(f"## {printer.name}")
-
-        # Connectivity
-        parts.append("### Connectivity")
-        if latest.wifi_signal_dbm is not None:
-            signal = latest.wifi_signal_dbm
-            quality = "Excellent" if signal > -50 else "Good" if signal > -60 else "Fair" if signal > -70 else "Poor"
-            parts.append(f"- WiFi: {signal} dBm ({quality})")
-        parts.append(f"- Last seen: {_local_dt(latest.timestamp, '%Y-%m-%d %H:%M:%S %Z')}")
-        age = (timezone.now() - latest.timestamp).total_seconds()
-        if age > 300:
-            parts.append(f"- **Warning**: No data for {_format_duration(age / 60)}")
-
-        # AMS environment
-        if latest.ams_humidity is not None or latest.ams_temp is not None:
-            parts.append("### AMS Environment")
-            if latest.ams_humidity is not None:
-                hum_status = "OK" if latest.ams_humidity < 5 else "High" if latest.ams_humidity < 8 else "Critical"
-                parts.append(f"- Humidity: {latest.ams_humidity} ({hum_status})")
-            if latest.ams_temp is not None:
-                parts.append(f"- Temperature: {latest.ams_temp}°C")
-
-        # HMS errors
-        if latest.hms:
-            parts.append("### Active HMS Alerts")
-            for msg in latest.hms:
-                parts.append(f"- {msg}")
-
-        # Recent failures
-        week_ago = timezone.now() - timedelta(days=7)
-        failed = PrintJob.objects.filter(
-            device=printer,
-            start_time__gte=week_ago,
-            final_status__in=["FAILED", "CANCELLED"],
-        )
-        if failed.exists():
-            parts.append(f"### Recent Failures (7d): {failed.count()}")
-            for job in failed.select_related("cloud_task")[:5]:
-                parts.append(f"- {_job_name(job)} ({job.final_status}) — {_local_dt(job.start_time, '%m-%d %H:%M')}")
-
-        # Success rate
-        week_jobs = PrintJob.objects.filter(device=printer, start_time__gte=week_ago)
-        total = week_jobs.count()
-        if total > 0:
-            success = week_jobs.filter(final_status="FINISH").count()
-            parts.append(f"\n**7-day success rate**: {success}/{total} ({100 * success // total}%)")
-
-        parts.append("")
-
-    return "\n".join(parts)
-
-
-def search_print_jobs(query):
-    """Search print jobs by project name or gcode file."""
-    from .models import PrintJob
-
-    if not query:
-        return "Please provide a search query."
-
-    jobs = PrintJob.objects.select_related("device", "cloud_task").filter(
-        Q(project_name__icontains=query)
-        | Q(gcode_file__icontains=query)
-        | Q(cloud_task__design_title__icontains=query)
-    )[:20]
-
-    if not jobs:
-        return f"No print jobs matching '{query}'."
-
-    lines = [f"# Search Results: '{query}'", ""]
-    lines.append(f"*{len(jobs)} results*\n")
-    lines.append("| ID | Project | Printer | Status | Date |")
-    lines.append("|----|---------|---------|--------|------|")
-    for j in jobs:
-        lines.append(
-            f"| {j.id} | {_job_name(j)} | {j.device.name} | "
-            f"{j.final_status or 'In Progress'} | {_local_dt(j.start_time, '%Y-%m-%d')} |"
-        )
-    return "\n".join(lines)
-
-
-def get_printing_summary(days=7):
-    """High-level activity summary."""
-    from .models import FilamentUsage, Printer, PrintJob
-
-    cutoff = timezone.now() - timedelta(days=int(days))
-    jobs = PrintJob.objects.filter(start_time__gte=cutoff)
-
-    total = jobs.count()
-    finished = jobs.filter(final_status="FINISH").count()
-    failed = jobs.filter(final_status="FAILED").count()
-    cancelled = jobs.filter(final_status="CANCELLED").count()
-    in_progress = jobs.filter(final_status__isnull=True).count()
-
-    total_minutes = jobs.filter(duration_minutes__isnull=False).aggregate(
-        total=Sum("duration_minutes")
-    )["total"] or 0
-
-    total_grams = FilamentUsage.objects.filter(
-        print_job__start_time__gte=cutoff,
-        consumed_grams__isnull=False,
-    ).aggregate(total=Sum("consumed_grams"))["total"] or 0
-
-    lines = [f"# Printing Summary (last {days} days)", ""]
-    lines.append(f"**Total Jobs**: {total}")
-    lines.append(f"- Completed: {finished}")
-    lines.append(f"- Failed: {failed}")
-    lines.append(f"- Cancelled: {cancelled}")
-    lines.append(f"- In Progress: {in_progress}")
-    if total > 0:
-        lines.append(f"- Success Rate: {100 * finished // total}%")
-    lines.append(f"\n**Total Print Time**: {_format_duration(total_minutes)}")
-    lines.append(f"**Total Filament Used**: {total_grams}g")
-
-    # Most printed projects
-    top_projects = (
-        jobs.values("project_name")
-        .annotate(count=Count("id"))
-        .order_by("-count")[:5]
-    )
-    if top_projects:
-        lines.append("\n### Most Printed")
-        for p in top_projects:
-            lines.append(f"- {p['project_name']} ({p['count']}x)")
-
-    # Active printers
-    active_printers = Printer.objects.filter(
-        print_jobs__start_time__gte=cutoff
-    ).distinct()
-    if active_printers.exists():
-        lines.append(f"\n**Active Printers**: {', '.join(p.name for p in active_printers)}")
-
-    return "\n".join(lines)
-
-
-def find_compatible_filament(type, min_remaining_percent=10, color=None):
-    """Find spools matching material type criteria."""
-    from .models import Filament
-
-    qs = Filament.objects.filter(
-        type__iexact=type,
-        remaining_percent__gte=int(min_remaining_percent),
-    )
-    if color:
-        qs = qs.filter(color__icontains=color)
-
-    filaments = qs[:20]
-    if not filaments:
-        return f"No {type} filament found with >={min_remaining_percent}% remaining."
-
-    lines = [f"# Compatible Filament: {type}", ""]
-    if color:
-        lines.append(f"*Color filter: {color}*\n")
-    lines.append(f"*{qs.count()} spools found*\n")
-    lines.append("| ID | Brand | Sub-type | Color | Remaining | In AMS |")
-    lines.append("|----|-------|----------|-------|-----------|--------|")
-    for f in filaments:
-        lines.append(
-            f"| {f.id} | {f.brand} | {f.sub_type or f.type} | "
-            f"{f.color} | {f.remaining_percent}% | "
-            f"{'Yes' if f.is_loaded_in_ams else 'No'} |"
-        )
-    return "\n".join(lines)
-
-
-# ─── Resources ───────────────────────────────────────────────────────────────
-
-
-def resource_printers():
-    """List all printers (resource)."""
-    return list_printers()
-
-
-def resource_printer_status(printer_id):
-    """Latest printer status (resource)."""
-    return get_printer_status(printer_id=printer_id)
-
-
-def resource_filaments():
-    """Full filament inventory (resource)."""
-    return list_filaments()
-
-
-def resource_filament_detail(filament_id):
-    """Single spool with usage (resource)."""
-    return get_filament_detail(filament_id=filament_id)
-
-
-def resource_recent_print_jobs():
-    """Last 20 print jobs (resource)."""
-    return get_print_history(limit=20)
-
-
-def resource_filament_types():
-    """Filament type registry (resource)."""
-    from .models import FilamentType
-
-    types = FilamentType.objects.all()
-    if not types.exists():
-        return "No filament types registered."
-
-    lines = ["# Filament Types", ""]
-    lines.append("| ID | Type | Sub-type | Brand |")
-    lines.append("|----|------|----------|-------|")
-    for t in types:
-        lines.append(f"| {t.id} | {t.type} | {t.sub_type or '-'} | {t.brand} |")
-    return "\n".join(lines)
-
-
-def resource_filament_colors():
-    """Filament color database (resource)."""
-    from .models import FilamentColor
-
-    colors = FilamentColor.objects.all()[:100]
-    if not colors:
-        return "No filament colors in database."
-
-    lines = ["# Filament Colors", ""]
-    lines.append(f"*Showing up to 100 of {FilamentColor.objects.count()}*\n")
-    lines.append("| Color | Hex | Type | Sub-type | Brand |")
-    lines.append("|-------|-----|------|----------|-------|")
-    for c in colors:
-        lines.append(
-            f"| {c.color_name} | #{c.color_code} | {c.filament_type} | "
-            f"{c.filament_sub_type or '-'} | {c.brand} |"
-        )
-    return "\n".join(lines)
-
-
-# ─── Prompts ─────────────────────────────────────────────────────────────────
-
-
-def prompt_printer_check_in(printer_id=None):
-    """Full status briefing: status + health + recent prints."""
-    parts = [
-        get_printer_status(printer_id=printer_id),
-        get_printer_health(printer_id=printer_id),
-        get_print_history(days=1, limit=5),
-    ]
-    return "\n\n---\n\n".join(parts)
-
-
-def prompt_filament_inventory_report():
-    """Inventory report with low-stock warnings."""
-    from .models import Filament
-
-    low_stock = Filament.objects.filter(remaining_percent__lte=20)
-    parts = [list_filaments()]
-    if low_stock.exists():
-        lines = ["\n## Low Stock Warnings"]
-        for f in low_stock:
-            lines.append(f"- **{f.brand} {f.type} {f.color}**: {f.remaining_percent}% remaining")
-        parts.append("\n".join(lines))
-    return "\n\n".join(parts)
-
-
-def prompt_print_job_review(job_id):
-    """Review a completed job."""
-    return get_print_job_detail(job_id)
-
-
-def prompt_weekly_digest():
-    """Weekly activity summary."""
-    parts = [
-        get_printing_summary(days=7),
-        get_filament_usage_stats(days=7, group_by="type"),
-    ]
-    return "\n\n---\n\n".join(parts)
-
-
-def prompt_troubleshoot_printer(printer_id=None):
-    """Diagnose issues from recent data."""
-    parts = [
-        get_printer_health(printer_id=printer_id),
-        get_printer_status(printer_id=printer_id),
-        get_temperature_history(printer_id=printer_id, hours=2),
-    ]
-    return "\n\n---\n\n".join(parts)

pyproject.toml

@@ -40,13 +40,18 @@ standalone = [
     "python-dotenv",
     "whitenoise",
 ]
+postgres = [
+    "psycopg2-binary",
+]
 mcp = [
-    "mcp[cli]>=1.0",
+    "mcp-gate @ git+https://github.com/RunLit/mcp-gate.git@v0.1.0",
+    "uvicorn",
 ]
 dev = [
     "ruff",
     "pytest",
     "pytest-django",
+    "anyio",
 ]
 
 [project.urls]

standalone/settings.py

@@ -63,16 +63,29 @@ TEMPLATES = [
 
 WSGI_APPLICATION = "standalone.wsgi.application"
 
-# Database — SQLite for zero-setup deployment
+# Database — SQLite for zero-setup deployment, or Postgres when DB_ENGINE=postgresql
+# is set (e.g. to point a standalone install at an existing database).
 DATA_DIR = Path(os.environ.get("DATA_DIR", BASE_DIR / "data"))
 DATA_DIR.mkdir(parents=True, exist_ok=True)
 
-DATABASES = {
-    "default": {
-        "ENGINE": "django.db.backends.sqlite3",
-        "NAME": DATA_DIR / "db.sqlite3",
+if os.environ.get("DB_ENGINE") == "postgresql":
+    DATABASES = {
+        "default": {
+            "ENGINE": "django.db.backends.postgresql",
+            "NAME": os.environ["DB_NAME"],
+            "USER": os.environ["DB_USER"],
+            "PASSWORD": os.environ["DB_PASSWORD"],
+            "HOST": os.environ["DB_HOST"],
+            "PORT": os.environ.get("DB_PORT", "5432"),
+        }
+    }
+else:
+    DATABASES = {
+        "default": {
+            "ENGINE": "django.db.backends.sqlite3",
+            "NAME": DATA_DIR / "db.sqlite3",
+        }
     }
-}
 
 # Password validation
 AUTH_PASSWORD_VALIDATORS = [

tests/conftest.py

@@ -0,0 +1,30 @@
+"""Suite-wide test fixtures.
+
+test_mcp_toolkit.py drives real MCP dispatch through `mcp.client.Client`,
+which routes each tool call's synchronous Django ORM work through
+`sync_to_async` onto its own dedicated worker thread rather than the test's
+own thread. pytest-django's `db` fixture isolates each test by wrapping it in
+an outer atomic transaction on *its* thread's connection and rolling that
+back afterwards -- but the worker thread opens a separate, thread-local
+connection that was never part of that transaction, so writes made through
+it commit for real and survive into later tests (e.g. Printer rows from
+test_mcp_toolkit.py leaking into test_resolve_printer_device.py's counts).
+
+This fixture flushes bambu_run's Printer table after each test in that file
+specifically, restoring isolation without touching the test file itself
+(which mirrors mcp-gate's proven end-to-end test pattern verbatim).
+"""
+from __future__ import annotations
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _isolate_async_mcp_dispatch_writes(request, django_db_blocker):
+    yield
+    if request.node.module.__name__ != "tests.test_mcp_toolkit":
+        return
+    with django_db_blocker.unblock():
+        from bambu_run.models import Printer
+
+        Printer.objects.all().delete()

tests/settings.py

@@ -1,4 +1,16 @@
 """Minimal Django settings for running bambu_run's pytest suite (in-memory SQLite)."""
+import os
+
+# test_mcp_toolkit.py drives real MCP dispatch through `mcp.client.Client`,
+# which runs the test body on the asyncio event loop (via pytest-anyio) while
+# still using plain synchronous Django ORM calls (Printer.objects.create(),
+# and the ORM calls inside the tool functions themselves). Django's asyncio
+# safety net treats any sync ORM call made while an event loop is running as
+# unsafe by default. It's genuinely safe here: sqlite is in-memory, tests run
+# single-threaded, and there's no concurrent request interleaving to corrupt
+# a shared connection. Set at settings-import time (before any test body
+# runs) rather than per-test, since Django reads this env var at call time.
+os.environ.setdefault("DJANGO_ALLOW_ASYNC_UNSAFE", "1")
 
 SECRET_KEY = "test-secret-key"
 

tests/test_mcp_filaments.py

@@ -0,0 +1,137 @@
+import pytest
+from django.utils import timezone
+
+from bambu_run.models import Filament, Printer, PrintJob, FilamentUsage
+from bambu_run.mcp.filaments import list_filaments, get_filament
+
+
+@pytest.mark.django_db
+def test_list_filaments_filters_material_sub_type_color():
+    Filament.objects.create(type="PLA", sub_type="PLA Matte", brand="Bambu Lab", color="Black", remaining_percent=50)
+    Filament.objects.create(type="PLA", sub_type="PLA Basic", brand="Bambu Lab", color="Black", remaining_percent=50)
+    Filament.objects.create(type="ABS", sub_type="ABS", brand="Bambu Lab", color="Black", remaining_percent=50)
+
+    result = list_filaments(material="PLA", sub_type="PLA Matte", color="black")
+
+    assert len(result["filaments"]) == 1
+    assert result["filaments"][0]["sub_type"] == "PLA Matte"
+    assert "generated_at" in result
+
+
+@pytest.mark.django_db
+def test_list_filaments_min_remaining_grams_excludes_unread_sentinel():
+    Filament.objects.create(type="ABS", brand="Bambu Lab", color="Mint", remaining_percent=-1, remaining_weight_grams=-10)
+    Filament.objects.create(type="ABS", brand="Bambu Lab", color="Orange", remaining_percent=68, remaining_weight_grams=680)
+
+    result = list_filaments(material="ABS", min_remaining_grams=250)
+
+    assert len(result["filaments"]) == 1
+    assert result["filaments"][0]["color_name"] == "Orange"
+
+
+@pytest.mark.django_db
+def test_list_filaments_min_remaining_grams_excludes_unread_even_with_low_threshold():
+    # The existing test only checks min_remaining_grams=250, which happens to
+    # exclude a -10g sentinel-derived value anyway. This test uses a threshold
+    # low enough that a naive filter would let the sentinel spool through.
+    Filament.objects.create(
+        type="ABS", brand="Bambu Lab", color="Mint",
+        remaining_percent=-1, remaining_weight_grams=-10,
+    )
+    Filament.objects.create(
+        type="ABS", brand="Bambu Lab", color="Orange",
+        remaining_percent=68, remaining_weight_grams=680,
+    )
+
+    result = list_filaments(material="ABS", min_remaining_grams=-20)
+
+    assert len(result["filaments"]) == 1
+    assert result["filaments"][0]["color_name"] == "Orange"
+
+
+@pytest.mark.django_db
+def test_list_filaments_unread_sentinel_status_not_low():
+    Filament.objects.create(type="ABS", brand="Bambu Lab", color="Mint", remaining_percent=-1)
+
+    result = list_filaments(material="ABS")
+
+    spool = result["filaments"][0]
+    assert spool["remaining_percent"] is None
+    assert spool["remaining_status"] == "unread"
+
+
+@pytest.mark.django_db
+def test_list_filaments_limit_hard_ceiling_100():
+    Filament.objects.bulk_create([
+        Filament(type="PLA", brand="Bambu Lab", color=f"Color {i}", remaining_percent=50)
+        for i in range(150)
+    ])
+
+    result = list_filaments(limit=1000)
+
+    assert len(result["filaments"]) == 100
+
+
+@pytest.mark.django_db
+def test_list_filaments_include_history_adds_recent_usage():
+    f = Filament.objects.create(type="PLA", brand="Bambu Lab", color="White", initial_weight_grams=1000, remaining_percent=80)
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    job = PrintJob.objects.create(device=p, project_name="Vase", start_time=timezone.now(), final_status="FINISH")
+    FilamentUsage.objects.create(print_job=job, filament=f, tray_id=0, starting_percent=90, ending_percent=80)
+
+    result = list_filaments(material="PLA", include_history=True)
+
+    history = result["filaments"][0]["recent_usage"]
+    assert history[0]["job_name"] == "Vase"
+    assert history[0]["consumed_grams"] == 100
+
+
+@pytest.mark.django_db
+def test_list_filaments_never_exposes_rfid_identifiers():
+    Filament.objects.create(
+        type="PLA", brand="Bambu Lab", color="White",
+        tray_uuid="SECRET-UUID", tag_uid="SECRET-TAG",
+    )
+
+    result = list_filaments(material="PLA")
+
+    entry_str = repr(result["filaments"][0])
+    assert "SECRET-UUID" not in entry_str
+    assert "SECRET-TAG" not in entry_str
+
+
+@pytest.mark.django_db
+def test_get_filament_returns_full_usage_history_with_data_quality():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    f = Filament.objects.create(type="ABS", brand="Bambu Lab", color="Black", initial_weight_grams=1000)
+    job1 = PrintJob.objects.create(device=p, project_name="Bracket", start_time=timezone.now(), final_status="FINISH")
+    job2 = PrintJob.objects.create(device=p, project_name="Vase", start_time=timezone.now(), final_status="FINISH")
+    FilamentUsage.objects.create(print_job=job1, filament=f, tray_id=0, starting_percent=80, ending_percent=70)
+    FilamentUsage.objects.create(print_job=job2, filament=f, tray_id=0, starting_percent=-1, ending_percent=100)
+
+    result = get_filament(f.id)
+
+    assert len(result["usage_history"]) == 2
+    reliable = next(u for u in result["usage_history"] if u["job_name"] == "Bracket")
+    unreliable = next(u for u in result["usage_history"] if u["job_name"] == "Vase")
+    assert reliable["data_quality"] == "ok"
+    assert reliable["consumed_grams"] == 100
+    assert unreliable["data_quality"] == "unreliable"
+    assert unreliable["consumed_grams"] is None
+    assert "generated_at" in result
+
+
+@pytest.mark.django_db
+def test_get_filament_never_exposes_purchase_price_or_cost():
+    f = Filament.objects.create(type="PLA", brand="Bambu Lab", color="White", purchase_price=25.50)
+
+    result = get_filament(f.id)
+
+    assert "purchase_price" not in result
+    assert "cost" not in result
+
+
+@pytest.mark.django_db
+def test_get_filament_unknown_id_raises():
+    with pytest.raises(Filament.DoesNotExist):
+        get_filament(999999)

tests/test_mcp_jobs.py

@@ -0,0 +1,196 @@
+import pytest
+from datetime import timedelta
+from django.utils import timezone
+
+from bambu_run.models import Printer, PrintJob, BambuCloudTask, Filament, FilamentUsage
+from bambu_run.mcp.jobs import list_print_jobs, get_print_job, get_print_statistics
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_orders_most_recent_first():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    older = PrintJob.objects.create(device=p, project_name="Old", start_time=now - timedelta(days=1), final_status="FINISH")
+    newer = PrintJob.objects.create(device=p, project_name="New", start_time=now, final_status="FINISH")
+
+    result = list_print_jobs()
+
+    assert [j["id"] for j in result["jobs"]] == [newer.id, older.id]
+    assert "generated_at" in result
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_filters_by_status_running():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    running = PrintJob.objects.create(device=p, project_name="Live", start_time=timezone.now(), final_status=None)
+    PrintJob.objects.create(device=p, project_name="Done", start_time=timezone.now(), final_status="FINISH")
+
+    result = list_print_jobs(status="running")
+
+    assert [j["id"] for j in result["jobs"]] == [running.id]
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_limit_hard_ceiling_100():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    PrintJob.objects.bulk_create([
+        PrintJob(device=p, project_name=f"Job {i}", start_time=now - timedelta(hours=i), final_status="FINISH")
+        for i in range(150)
+    ])
+
+    result = list_print_jobs(limit=1000)
+
+    assert len(result["jobs"]) == 100
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_prefers_cloud_weight_and_tags_source():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    cloud = BambuCloudTask.objects.create(task_id=1, design_title="Cool Bracket", weight_grams=70.10)
+    PrintJob.objects.create(
+        device=p, project_name="raw_name", start_time=timezone.now(),
+        final_status="FINISH", cloud_task=cloud,
+    )
+
+    result = list_print_jobs()
+
+    job = result["jobs"][0]
+    assert job["display_name"] == "Cool Bracket"
+    assert job["filament_grams"] == 70.10
+    assert job["filament_grams_source"] == "cloud"
+    assert job["cost"] is None
+
+
+@pytest.mark.django_db
+def test_list_print_jobs_display_name_falls_back_to_project_name():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrintJob.objects.create(device=p, project_name="raw_project", start_time=timezone.now(), final_status="FINISH")
+
+    result = list_print_jobs()
+
+    assert result["jobs"][0]["display_name"] == "raw_project"
+
+
+@pytest.mark.django_db
+def test_get_print_job_marks_negative_consumption_unreliable():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    job = PrintJob.objects.create(device=p, project_name="Bracket", start_time=timezone.now(), final_status="FINISH")
+    filament = Filament.objects.create(type="ABS", brand="Bambu Lab", color="Mint", initial_weight_grams=1000)
+    FilamentUsage.objects.create(
+        print_job=job, filament=filament, tray_id=0,
+        starting_percent=51, ending_percent=52,  # ordinary RFID jitter, spec finding
+    )
+
+    result = get_print_job(job.id)
+
+    usage = result["filament_usages"][0]
+    assert usage["consumed_grams"] is None
+    assert usage["data_quality"] == "unreliable"
+    assert "generated_at" in result
+
+
+@pytest.mark.django_db
+def test_get_print_job_computes_estimate_delta_percent():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    cloud = BambuCloudTask.objects.create(task_id=1, cost_time_seconds=205 * 60)
+    job = PrintJob.objects.create(
+        device=p, project_name="Bracket", start_time=timezone.now(),
+        final_status="FINISH", cloud_task=cloud, duration_minutes=237,
+    )
+
+    result = get_print_job(job.id)
+
+    assert result["estimate_delta_percent"] == round(100 * (237 - 205) / 205, 1)
+
+
+@pytest.mark.django_db
+def test_get_print_job_unknown_id_raises():
+    with pytest.raises(PrintJob.DoesNotExist):
+        get_print_job(999999)
+
+
+@pytest.mark.django_db
+def test_get_print_statistics_success_rate_and_status_counts():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    PrintJob.objects.create(device=p, project_name="A", start_time=now, final_status="FINISH")
+    PrintJob.objects.create(device=p, project_name="B", start_time=now, final_status="FAILED")
+
+    result = get_print_statistics(days=30)
+
+    assert result["total_jobs"] == 2
+    assert result["success_rate"] == 0.5
+    assert result["status_counts"]["FINISH"] == 1
+    assert result["status_counts"]["FAILED"] == 1
+    assert "generated_at" in result
+
+
+@pytest.mark.django_db
+def test_get_print_statistics_excludes_ending_sentinel_from_material_mix():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    job = PrintJob.objects.create(device=p, project_name="A", start_time=timezone.now(), final_status="FINISH")
+    pla = Filament.objects.create(type="PLA", brand="Bambu Lab", color="Black", initial_weight_grams=1000)
+    # ending_percent=-1 (no RFID read at job end) must not be counted as 610g of consumption
+    FilamentUsage.objects.create(print_job=job, filament=pla, tray_id=0, starting_percent=60, ending_percent=-1)
+
+    result = get_print_statistics(days=30)
+
+    assert result["material_mix_grams"] == {}
+    assert result["most_used_material"] is None
+
+
+@pytest.mark.django_db
+def test_get_print_statistics_material_mix_excludes_negative_consumption():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    job = PrintJob.objects.create(device=p, project_name="A", start_time=timezone.now(), final_status="FINISH")
+    pla = Filament.objects.create(type="PLA", brand="Bambu Lab", color="Black", initial_weight_grams=1000)
+    abs_f = Filament.objects.create(type="ABS", brand="Bambu Lab", color="White", initial_weight_grams=1000)
+    FilamentUsage.objects.create(print_job=job, filament=pla, tray_id=0, starting_percent=74, ending_percent=68)  # 6% -> 60g, valid
+    FilamentUsage.objects.create(print_job=job, filament=abs_f, tray_id=1, starting_percent=51, ending_percent=52)  # jitter, excluded
+
+    result = get_print_statistics(days=30)
+
+    assert result["material_mix_grams"] == {"PLA": 60.0}
+    assert result["most_used_material"] == "PLA"
+
+
+@pytest.mark.django_db
+def test_get_print_statistics_estimate_accuracy_mean_and_median():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    cloud1 = BambuCloudTask.objects.create(task_id=1, cost_time_seconds=100 * 60)
+    cloud2 = BambuCloudTask.objects.create(task_id=2, cost_time_seconds=100 * 60)
+    PrintJob.objects.create(
+        device=p, project_name="A", start_time=timezone.now(),
+        final_status="FINISH", cloud_task=cloud1, duration_minutes=110,
+    )
+    PrintJob.objects.create(
+        device=p, project_name="B", start_time=timezone.now(),
+        final_status="FINISH", cloud_task=cloud2, duration_minutes=130,
+    )
+
+    result = get_print_statistics(days=30)
+
+    assert result["estimate_accuracy"]["mean_ratio"] == 1.2
+    assert result["estimate_accuracy"]["median_ratio"] == 1.2
+
+
+@pytest.mark.django_db
+def test_get_print_statistics_cost_always_null_and_unavailable():
+    result = get_print_statistics(days=30)
+    assert result["cost_total"] is None
+    assert result["cost_available"] is False
+
+
+@pytest.mark.django_db
+def test_get_print_statistics_failures_lists_top_failing_projects():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    PrintJob.objects.create(device=p, project_name="Flaky", start_time=now, final_status="FAILED")
+    PrintJob.objects.create(device=p, project_name="Flaky", start_time=now, final_status="FAILED")
+    PrintJob.objects.create(device=p, project_name="Solid", start_time=now, final_status="FINISH")
+
+    result = get_print_statistics(days=30)
+
+    assert result["failures"]["count"] == 2
+    assert result["failures"]["top_failing_projects"][0] == {"name": "Flaky", "count": 2}

tests/test_mcp_server_command.py

@@ -0,0 +1,24 @@
+import pytest
+from mcp_gate import OpenIdentity, StaticTokenIdentity
+
+from bambu_run.management.commands.bambu_mcp_server import ALL_CAPABILITIES, _build_gate
+from bambu_run.mcp.toolkit import build_toolkit
+
+
+def test_build_gate_stdio_uses_open_identity_with_all_capabilities():
+    gate = _build_gate(build_toolkit(), transport="stdio")
+    assert isinstance(gate.identity, OpenIdentity)
+
+
+def test_build_gate_http_requires_api_key():
+    with pytest.raises(ValueError, match="BAMBU_RUN_MCP_API_KEY"):
+        _build_gate(build_toolkit(), transport="http", api_key=None)
+
+
+def test_build_gate_http_with_key_uses_static_token_identity():
+    gate = _build_gate(build_toolkit(), transport="http", api_key="secret123")
+    assert isinstance(gate.identity, StaticTokenIdentity)
+
+
+def test_all_capabilities_matches_the_three_spec_capabilities():
+    assert set(ALL_CAPABILITIES) == {"read_printer", "read_print_history", "read_filament"}

tests/test_mcp_shapes.py

@@ -0,0 +1,140 @@
+import pytest
+from datetime import timedelta
+from django.utils import timezone
+
+from bambu_run.models import Printer
+from bambu_run.mcp.shapes import (
+    STALE_THRESHOLD_SECONDS,
+    freshness,
+    remaining_status,
+    smoothed_remaining_percent,
+    classify_alerts,
+    display_name,
+    normalize_consumed,
+    resolve_printer,
+)
+
+
+def test_freshness_fresh_reading():
+    now = timezone.now()
+    ts = now - timedelta(seconds=10)
+    assert freshness(ts, now) == {"data_age_seconds": 10, "stale": False}
+
+
+def test_freshness_stale_reading():
+    now = timezone.now()
+    ts = now - timedelta(seconds=STALE_THRESHOLD_SECONDS + 30)
+    result = freshness(ts, now)
+    assert result["stale"] is True
+
+
+def test_remaining_status_unread_sentinel():
+    assert remaining_status(-1) == (None, "unread")
+
+
+def test_remaining_status_none_is_unread():
+    assert remaining_status(None) == (None, "unread")
+
+
+def test_remaining_status_empty():
+    assert remaining_status(0) == (0, "empty")
+
+
+def test_remaining_status_low_boundary():
+    assert remaining_status(20) == (20, "low")
+    assert remaining_status(21) == (21, "ok")
+
+
+def test_remaining_status_ok():
+    assert remaining_status(72) == (72, "ok")
+
+
+def test_smoothed_remaining_percent_rejects_spikes():
+    # Real data pattern (spec, job 122): genuine values bouncing 22-35%.
+    readings = [35, 22, 30, 28, 25]
+    assert smoothed_remaining_percent(readings) == (28, "ok")
+
+
+def test_smoothed_remaining_percent_excludes_unread_sentinel():
+    readings = [-1, -1, 90, 88, 91]
+    assert smoothed_remaining_percent(readings) == (90, "ok")
+
+
+def test_smoothed_remaining_percent_all_unread():
+    assert smoothed_remaining_percent([-1, -1, -1]) == (None, "unread")
+
+
+def test_smoothed_remaining_percent_only_uses_5_most_recent():
+    readings = [10, 10, 10, 10, 10, 100]
+    value, _ = smoothed_remaining_percent(readings)
+    assert value == 10
+
+
+def test_classify_alerts_persistent_above_20_percent():
+    counts = {(83887616, 131078): 25}
+    assert classify_alerts(counts, total_rows=100) == [{
+        "attr": 83887616, "code": 131078,
+        "kind": "persistent", "occurrences": 25, "severity": "unknown",
+    }]
+
+
+def test_classify_alerts_transient_below_5_percent():
+    counts = {(50336256, 131073): 3}
+    alerts = classify_alerts(counts, total_rows=100)
+    assert alerts[0]["kind"] == "transient"
+
+
+def test_classify_alerts_omits_ambiguous_band():
+    counts = {(1, 2): 10}  # exactly 10% of 100 rows: neither >20% nor <5%
+    assert classify_alerts(counts, total_rows=100) == []
+
+
+def test_display_name_falls_back_through_chain():
+    assert display_name("", "", "raw_project") == "raw_project"
+    assert display_name("", "plate_1", "raw_project") == "plate_1"
+    assert display_name("Cool Design", "plate_1", "raw_project") == "Cool Design"
+
+
+def test_normalize_consumed_ordinary_reading():
+    assert normalize_consumed(74, 68, initial_weight_grams=1000) == (6, 60, "ok")
+
+
+def test_normalize_consumed_rfid_jitter_goes_unreliable_not_negative():
+    # spec finding: start=51% end=52% -> raw consumed=-1% (ordinary jitter)
+    assert normalize_consumed(51, 52, initial_weight_grams=1000) == (None, None, "unreliable")
+
+
+def test_normalize_consumed_sentinel_start_goes_unreliable():
+    # spec finding: jobs 62/63/70/77, start=-1% end=100%
+    assert normalize_consumed(-1, 100, initial_weight_grams=1000) == (None, None, "unreliable")
+
+
+def test_normalize_consumed_sentinel_end_goes_unreliable():
+    # symmetric case: ending_percent == -1 (no RFID reading at job end)
+    # must not fabricate a plausible-looking consumption figure
+    assert normalize_consumed(74, -1, initial_weight_grams=1000) == (None, None, "unreliable")
+
+
+@pytest.mark.django_db
+def test_resolve_printer_by_id():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    assert list(resolve_printer(p.id)) == [p]
+
+
+@pytest.mark.django_db
+def test_resolve_printer_by_name():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    assert list(resolve_printer("RNL-H2C")) == [p]
+
+
+@pytest.mark.django_db
+def test_resolve_printer_none_returns_all_active():
+    p1 = Printer.objects.create(name="A", model="H2C", is_active=True)
+    Printer.objects.create(name="B", model="X1C", is_active=False)
+    assert list(resolve_printer(None)) == [p1]
+
+
+@pytest.mark.django_db
+def test_resolve_printer_unknown_name_raises():
+    with pytest.raises(Printer.DoesNotExist):
+        resolve_printer("nonexistent")

tests/test_mcp_status.py

@@ -0,0 +1,283 @@
+import pytest
+from datetime import timedelta
+from django.utils import timezone
+
+from bambu_run.models import Printer, PrinterMetrics
+from bambu_run.mcp.status import get_printer_status
+
+
+@pytest.mark.django_db
+def test_get_printer_status_no_data_yet():
+    Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+
+    result = get_printer_status()
+
+    assert len(result["printers"]) == 1
+    entry = result["printers"][0]
+    assert entry["state"] == "no_data"
+    assert entry["stale"] is True
+    assert entry["job"] is None
+
+
+@pytest.mark.django_db
+def test_get_printer_status_running_job_computes_finish_time():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now(), gcode_state="RUNNING",
+        print_percent=63, layer_num=189, total_layer_num=301,
+        remaining_time_min=47, subtask_name="SyncGen LightQube Bracket",
+    )
+
+    result = get_printer_status(printer=p.id)
+
+    job = result["printers"][0]["job"]
+    assert job["print_percent"] == 63
+    assert job["name"] == "SyncGen LightQube Bracket"
+    assert job["estimated_finish_at"] is not None
+
+
+@pytest.mark.django_db
+def test_get_printer_status_idle_has_no_job():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(device=p, timestamp=timezone.now(), gcode_state="FINISH")
+
+    result = get_printer_status(printer=p.id)
+
+    assert result["printers"][0]["job"] is None
+
+
+@pytest.mark.django_db
+def test_get_printer_status_stale_when_collector_gap_exceeds_120s():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now() - timedelta(seconds=200), gcode_state="RUNNING",
+    )
+
+    result = get_printer_status(printer=p.id)
+
+    assert result["printers"][0]["stale"] is True
+
+
+@pytest.mark.django_db
+def test_get_printer_status_ams_slot_carries_smoothed_percent_and_status():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    # Same tray, jittering readings across the 3 most recent polls (oldest first).
+    for i, pct in enumerate([22, 30, 35]):
+        PrinterMetrics.objects.create(
+            device=p, timestamp=now - timedelta(seconds=(2 - i) * 30), gcode_state="RUNNING",
+            filaments=[{
+                "ams_unit_id": 1, "tray_id": "2", "slot": "A01-P3",
+                "type": "PLA", "sub_type": "PLA Matte", "color": "E8AFCFFF",
+                "remain_percent": pct,
+            }],
+        )
+
+    result = get_printer_status(printer=p.id)
+
+    unit = result["printers"][0]["ams_units"][0]
+    slot = unit["slots"][0]
+    assert slot["remain_percent"] == 30  # median of [35, 30, 22] most-recent-first
+    assert slot["remain_status"] == "ok"
+    assert slot["color_hex"] == "#E8AFCF"
+
+
+@pytest.mark.django_db
+def test_get_printer_status_ams_slot_unread_sentinel():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now(), gcode_state="RUNNING",
+        filaments=[{"ams_unit_id": 0, "tray_id": "0", "type": "ABS", "remain_percent": -1}],
+    )
+
+    result = get_printer_status(printer=p.id)
+
+    slot = result["printers"][0]["ams_units"][0]["slots"][0]
+    assert slot["remain_percent"] is None
+    assert slot["remain_status"] == "unread"
+
+
+@pytest.mark.django_db
+def test_get_printer_status_ams_units_have_independent_humidity():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now(), gcode_state="RUNNING",
+        ams_humidity_raw=4,  # unit-0-only flat column, must NOT be used for unit 1
+        ams_units=[
+            {"unit_id": 0, "ams_type": "AMS 2 Pro", "humidity_raw": 4, "temp": 28.0},
+            {"unit_id": 1, "ams_type": "AMS", "humidity_raw": 7, "temp": 24.0},
+        ],
+        filaments=[
+            {"ams_unit_id": 0, "tray_id": "0", "type": "ABS", "remain_percent": 50},
+            {"ams_unit_id": 1, "tray_id": "0", "type": "PLA", "remain_percent": 60},
+        ],
+    )
+
+    result = get_printer_status(printer=p.id)
+
+    units_by_id = {u["unit_id"]: u for u in result["printers"][0]["ams_units"]}
+    assert units_by_id[0]["humidity_raw"] == 4
+    assert units_by_id[1]["humidity_raw"] == 7
+    assert units_by_id[0]["temp_c"] == 28.0
+    assert units_by_id[1]["temp_c"] == 24.0
+
+
+@pytest.mark.django_db
+def test_get_printer_status_tolerates_malformed_tray_id():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(
+        device=p, timestamp=timezone.now(), gcode_state="RUNNING",
+        filaments=[{"ams_unit_id": 0, "tray_id": None, "type": "ABS", "remain_percent": 50}],
+    )
+
+    result = get_printer_status(printer=p.id)  # must not raise
+
+    assert result["printers"][0]["ams_units"][0]["slots"][0]["slot"] == -1
+
+
+@pytest.mark.django_db
+def test_get_printer_status_never_exposes_serial_or_ip():
+    p = Printer.objects.create(
+        name="RNL-H2C", model="H2C", is_active=True,
+        serial_number="SECRET123", ip_address="192.168.1.50",
+    )
+    PrinterMetrics.objects.create(device=p, timestamp=timezone.now())
+
+    result = get_printer_status(printer=p.id)
+
+    entry_str = repr(result["printers"][0])
+    assert "SECRET123" not in entry_str
+    assert "192.168.1.50" not in entry_str
+
+
+@pytest.mark.django_db
+def test_get_printer_status_alerts_use_frequency_heuristic_not_hms_emptiness():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    # 1 of 5 rows in the alert window carries an HMS code -- below the 5%
+    # "transient" floor at this row count is not meaningful, but the point
+    # of this test is that a single non-empty hms row does NOT by itself
+    # produce an alert with unconditional certainty; it is classified, not
+    # surfaced verbatim. With 1/5 = 20%, exactly at the persistent boundary
+    # (not > 20%), so it lands in the omitted band.
+    for i in range(5):
+        hms = [{"attr": 1, "code": 2}] if i == 0 else []
+        PrinterMetrics.objects.create(device=p, timestamp=now - timedelta(minutes=i), hms=hms)
+
+    result = get_printer_status(printer=p.id)
+
+    assert result["printers"][0]["alerts"] == []
+
+
+from bambu_run.mcp.status import get_metrics_history
+
+
+@pytest.mark.django_db
+def test_get_metrics_history_returns_series_within_window():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    for i in range(3):
+        PrinterMetrics.objects.create(
+            device=p, timestamp=now - timedelta(hours=i),
+            nozzle_temp=200 + i, bed_temp=60, chamber_temp=35,
+            print_percent=50 + i, ams_humidity_raw=5, cooling_fan_speed=100,
+        )
+    PrinterMetrics.objects.create(device=p, timestamp=now - timedelta(hours=10), nozzle_temp=999)
+
+    result = get_metrics_history(printer=p.id, hours=6)
+
+    assert result["printer"]["id"] == p.id
+    assert len(result["timestamps"]) == 3
+    assert 999.0 not in result["series"]["nozzle_c"]
+    assert "generated_at" in result
+
+
+@pytest.mark.django_db
+def test_get_metrics_history_filters_to_requested_metrics():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(device=p, timestamp=timezone.now(), nozzle_temp=200, print_percent=10)
+
+    result = get_metrics_history(printer=p.id, metrics=["progress"])
+
+    assert "print_percent" in result["series"]
+    assert "nozzle_c" not in result["series"]
+
+
+@pytest.mark.django_db
+def test_get_metrics_history_hard_ceiling_500_points():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    PrinterMetrics.objects.bulk_create([
+        PrinterMetrics(device=p, timestamp=now - timedelta(minutes=i))
+        for i in range(600)
+    ])
+
+    result = get_metrics_history(printer=p.id, hours=24, max_points=10000)
+
+    assert len(result["timestamps"]) <= 500
+
+
+@pytest.mark.django_db
+def test_get_metrics_history_no_matching_printer_returns_empty_shape():
+    result = get_metrics_history(printer=None)
+    assert result["printer"] is None
+    assert result["timestamps"] == []
+    assert result["series"] == {}
+    assert "generated_at" in result
+
+
+from bambu_run.models import Hotend
+from bambu_run.mcp.status import get_hardware_health
+
+
+@pytest.mark.django_db
+def test_get_hardware_health_reports_wifi_and_humidity_averages():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    PrinterMetrics.objects.create(device=p, timestamp=now, wifi_signal_dbm=-50, ams_humidity_raw=4)
+    PrinterMetrics.objects.create(device=p, timestamp=now - timedelta(hours=1), wifi_signal_dbm=-60, ams_humidity_raw=6)
+
+    result = get_hardware_health(printer=p.id, days=7)
+
+    assert result["wifi_signal_dbm"]["current"] == -50
+    assert result["wifi_signal_dbm"]["window_avg"] == -55.0
+    assert result["ams_humidity_raw_window_avg"] == 5.0
+
+
+@pytest.mark.django_db
+def test_get_hardware_health_excludes_wear_percent_field():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    PrinterMetrics.objects.create(device=p, timestamp=timezone.now())
+    Hotend.objects.create(
+        printer=p, serial_number="SN1", raw_id=0, is_toolhead=True,
+        used_time_seconds=3600, wear_percent=100,
+    )
+
+    result = get_hardware_health(printer=p.id)
+
+    hotend = result["hotends"][0]
+    assert hotend["used_time_hours"] == 1.0
+    assert "wear_percent" not in hotend
+
+
+@pytest.mark.django_db
+def test_get_hardware_health_classifies_persistent_alert():
+    p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    now = timezone.now()
+    for i in range(30):
+        PrinterMetrics.objects.create(
+            device=p, timestamp=now - timedelta(minutes=i),
+            hms=[{"attr": 83887616, "code": 131078}],
+        )
+    for i in range(30, 100):
+        PrinterMetrics.objects.create(device=p, timestamp=now - timedelta(minutes=i), hms=[])
+
+    result = get_hardware_health(printer=p.id, days=1)
+
+    assert result["alerts"][0]["kind"] == "persistent"
+    assert result["alerts"][0]["occurrences"] == 30
+
+
+@pytest.mark.django_db
+def test_get_hardware_health_no_printer_returns_empty_shape():
+    assert get_hardware_health(printer=None) == {"printer": None}

tests/test_mcp_toolkit.py

@@ -0,0 +1,84 @@
+"""Proves this toolkit's capability assignments are wired correctly through
+real MCP dispatch -- not by calling tool functions directly. See
+~/src/mcp-gate/tests/test_end_to_end.py for the pattern this is modeled on;
+mcp-gate's own HTTP-shaped dispatch machinery is already tested there, so
+this file only needs the stdio-shaped Client, which is enough to drive
+server.middleware and prove which tools this toolkit's capability tags
+actually gate.
+"""
+import pytest
+from mcp.client import Client
+from mcp.shared.exceptions import MCPError
+
+from mcp_gate import Gate, OpenIdentity
+
+from bambu_run.mcp.toolkit import build_toolkit
+from bambu_run.models import Printer
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"
+
+
+async def test_full_capability_caller_sees_all_eight_tools(db):
+    Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    gate = Gate(
+        name="test",
+        toolkits=[build_toolkit()],
+        identity=OpenIdentity(["read_printer", "read_print_history", "read_filament"]),
+    )
+    server = gate.build_server()
+
+    async with Client(server, mode="legacy") as client:
+        result = await client.list_tools()
+
+    names = {t.name for t in result.tools}
+    assert names == {
+        "get_printer_status", "get_metrics_history", "get_hardware_health",
+        "list_print_jobs", "get_print_job", "get_print_statistics",
+        "list_filaments", "get_filament",
+    }
+
+
+async def test_read_printer_only_caller_cannot_see_or_call_filament_tool(db):
+    Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    gate = Gate(
+        name="test",
+        toolkits=[build_toolkit()],
+        identity=OpenIdentity(["read_printer"]),
+    )
+    server = gate.build_server()
+
+    async with Client(server, mode="legacy") as client:
+        visible = await client.list_tools()
+        visible_names = {t.name for t in visible.tools}
+        assert "list_filaments" not in visible_names
+        assert "list_print_jobs" not in visible_names
+        assert "get_printer_status" in visible_names
+
+        with pytest.raises(MCPError):
+            await client.call_tool("list_filaments", {})
+
+
+async def test_read_printer_caller_can_call_get_printer_status(db):
+    Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True)
+    gate = Gate(
+        name="test",
+        toolkits=[build_toolkit()],
+        identity=OpenIdentity(["read_printer"]),
+    )
+    server = gate.build_server()
+
+    async with Client(server, mode="legacy") as client:
+        result = await client.call_tool("get_printer_status", {})
+        assert result is not None
+
+
+def test_build_toolkit_has_no_name_collisions():
+    # Toolkit.tool() raises ValueError on a name collision at registration
+    # time -- building it twice without raising is itself the assertion.
+    build_toolkit()
+    build_toolkit()