Source code for src.agents.prusa_agent

"""
Prusa 3D Printer agent with MCP server integration.

This agent specializes in interacting with Prusa Connect to manage
3D printers, monitor print jobs, and control printer operations.

It was the one agent that reimplemented ``BaseAgent``'s graph instead of
inheriting it. That divergence cost more than it bought -- every guard added to
``BaseAgent`` had to be duplicated here by hand, and the two it had accumulated
by the end (empty-tools routing and hallucinated-call stripping) were
line-for-line copies of the originals.
"""

import logging
import os

from src.agents.base_agent import BaseAgent
from src.tools.mcp_client import MCPSSEClient
from src.tools.mcp_tools import load_mcp_tools
from src.tools.print_confirmation import guard
from src.utils.prompts import PRUSA_AGENT_SYSTEM_PROMPT

logger = logging.getLogger(__name__)

_ADMIN_ONLY_TOOLS = frozenset({"connect_login"})




class PrusaAgent(BaseAgent):
    """Agent specialized in 3D printer management via Prusa Connect."""



    def __init__(
        self,
        model_name: str | None = None,
        skip_mcp: bool = False,
        temperature: float | None = None,
        seed: int | None = None,
    ):
        """Initialize the Prusa agent.

        Args:
            model_name: Name of the LLM model to use (defaults to config.llm_model)
            skip_mcp: Skip MCP server initialization (useful for testing). Also
                honoured via the SKIP_MCP environment variable.
            temperature: Model temperature (defaults to config.llm_temperature)
            seed: Random seed for model (defaults to config.llm_seed)
        """
        skip_mcp = skip_mcp or os.getenv("SKIP_MCP", "false").lower() == "true"
        self.mcp_server_url = os.getenv("PRUSA_MCP_URL", "http://localhost:8765")
        # Must exist before super().__init__, which calls _create_tools().
        self.mcp_client = (
            None if skip_mcp else MCPSSEClient(server_url=self.mcp_server_url)
        )
        super().__init__(model_name=model_name, temperature=temperature, seed=seed)


    def _create_tools(self) -> list:
        """Load the printer tools from the MCP server.

        A server that is not there must not take the app down. The supervisor
        builds every agent in its own ``__init__``, so an exception here surfaces
        as "the assistant failed to start" -- with a printer stack trace,
        whatever the user was actually trying to do. That became reachable in
        ordinary use once SKIP_MCP turned into a master switch over the
        whiteboard as well: enabling the canvas enables this agent too, so
        running the canvas server without the printer server crashed startup.
        """
        if not self.mcp_client:
            return []
        try:
            # Every tool the server advertises is exposed automatically, so a
            # tool added server-side is callable here without any change in this
            # repo. `guard` is what keeps that from meaning a chat message can
            # start a print: it refuses the hardware-moving tools until a human
            # has approved that exact call.
            tools = load_mcp_tools(self.mcp_client, gate=guard)
            return [tool for tool in tools if tool.name not in _ADMIN_ONLY_TOOLS]
        except Exception:
            logger.warning(
                "Prusa MCP server unreachable at %s; the printer agent will "
                "have no tools",
                self.mcp_server_url,
                exc_info=True,
            )
            return []

    def _get_system_prompt(self) -> str:
        """Get the system prompt for the Prusa agent.

        Returns:
            System prompt string
        """
        return PRUSA_AGENT_SYSTEM_PROMPT