Source code for src.agents.excalidraw_agent

"""
Whiteboard agent, backed by the canvas MCP server.

Draws on the Excalidraw component the user is looking at, and reads back what
is on it. The scene itself lives in the canvas MCP server
(``services/excalidraw_mcp_server/``), keyed by conversation.
"""

import logging

from config import config
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.utils.prompts import EXCALIDRAW_AGENT_SYSTEM_PROMPT

logger = logging.getLogger(__name__)




class ExcalidrawAgent(BaseAgent):
    """Agent specialized in drawing on the user's whiteboard."""



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

        Args:
            model_name: Name of the LLM model to use (defaults to config.llm_model)
            temperature: Model temperature (defaults to config.llm_temperature)
            seed: Random seed for model (defaults to config.llm_seed)
            skip_canvas: Skip the MCP connection (useful for testing). Also
                honoured via SKIP_CANVAS / SKIP_MCP -- see ``config.canvas_enabled``.
        """
        # Both must exist before super().__init__, which calls _create_tools().
        self.canvas_id: str | None = None
        self.mcp_client = (
            None
            if (skip_canvas or not config.canvas_enabled)
            else MCPSSEClient(server_url=config.canvas_mcp_url)
        )
        super().__init__(model_name=model_name, temperature=temperature, seed=seed)


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

        ``canvas_id`` and ``source`` are injected rather than exposed: the model
        cannot know a conversation's UUID, and the injected values win over
        anything it supplies, so it can neither write to another conversation's
        whiteboard nor claim the user synced a drawing they did not. The callable
        is re-read per call, so a chat switch between turns is picked up without
        rebuilding the tools.

        A canvas server that is down must not take the whole app with it: the
        supervisor builds every agent in its own ``__init__``, so an exception
        here would surface as "the assistant failed to start".
        """
        if not self.mcp_client:
            return []
        try:
            return load_mcp_tools(
                self.mcp_client,
                hidden_args=lambda: {
                    "canvas_id": self.canvas_id or "default",
                    "source": "agent",
                },
            )
        except Exception:
            logger.warning(
                "Canvas MCP server unreachable at %s; the whiteboard agent will "
                "have no tools",
                config.canvas_mcp_url,
                exc_info=True,
            )
            return []

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

        Returns:
            System prompt string
        """
        return EXCALIDRAW_AGENT_SYSTEM_PROMPT