ben/Bambu-Run
1
0
Fork 0
mirror of https://github.com/RunLit/Bambu-Run.git synced 2026-10-21 15:37:38 +01:00
Code Issues Packages Projects Releases Wiki Activity

feat(mcp): serve the toolkit via mcp-gate; retire mcp_tools.py and the old FastMCP-based command

Browse Source
This commit is contained in:
RNL
2026-08-19 22:30:52 +10:00
parent 10034a2de4
commit e23650eaa4
3 changed files with 101 additions and 1050 deletions
Download Patch File Download Diff File Expand all files Collapse all files

393
bambu_run/management/commands/bambu_mcp_server.py
View File

@@ -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.
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)
@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"
)
raise CommandError("uvicorn is required for HTTP 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()
asgi_kwargs = {}
if options["allowed_host"]:
from mcp.server.transport_security import TransportSecuritySettings
asgi_kwargs["transport_security"] = TransportSecuritySettings(allowed_hosts=options["allowed_host"])
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)
else:
self.stdout.write(
self.style.SUCCESS("Starting Bambu-Run MCP server (stdio)")
)
mcp.run(transport="stdio")

728
bambu_run/mcp_tools.py
View File

@@ -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)

24
tests/test_mcp_server_command.py Normal file
View File

@@ -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"}