Timmy-time-dashboard/src/timmy/mcp_tools.py

"""MCP tool server factories for Agno agent integration.

Provides factory functions that create ``MCPTools`` instances for external
tool servers (Gitea, Filesystem) using stdio transport.  Also provides a
standalone async helper for filing Gitea issues from the thinking engine
without going through the full LLM loop.

Usage::

    from timmy.mcp_tools import create_gitea_mcp_tools, create_filesystem_mcp_tools

    # In agent creation (added to tools list):
    gitea_tools = create_gitea_mcp_tools()
    fs_tools = create_filesystem_mcp_tools()

    # Direct issue filing (thinking engine):
    from timmy.mcp_tools import create_gitea_issue_via_mcp
    result = await create_gitea_issue_via_mcp("Bug title", "Body", "bug")
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from PIL import ImageDraw
import os
import shutil
import sqlite3
import uuid
from contextlib import closing
from datetime import datetime
from pathlib import Path

import httpx

from config import settings

logger = logging.getLogger(__name__)

# Module-level cache for the standalone issue-filing session
_issue_session = None


def _parse_command(command_str: str) -> tuple[str, list[str]]:
    """Split a command string into (executable, args).

    Handles ``~/`` expansion and resolves via PATH if needed.
    E.g. ``"gitea-mcp-server -t stdio"`` → ``("/opt/homebrew/bin/gitea-mcp-server", ["-t", "stdio"])``
    """
    parts = command_str.split()
    executable = os.path.expanduser(parts[0])

    # If not an absolute path, resolve via shutil.which
    if not os.path.isabs(executable):
        resolved = shutil.which(executable)
        if resolved:
            executable = resolved
        else:
            # Check common binary locations not always on PATH
            for candidate_dir in ["~/go/bin", "~/.local/bin", "~/bin"]:
                candidate = os.path.expanduser(os.path.join(candidate_dir, parts[0]))
                if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
                    executable = candidate
                    break

    return executable, parts[1:]


def _gitea_server_params():
    """Build ``StdioServerParameters`` for the Gitea MCP server."""
    from mcp.client.stdio import StdioServerParameters

    exe, args = _parse_command(settings.mcp_gitea_command)
    return StdioServerParameters(
        command=exe,
        args=args,
        env={
            "GITEA_ACCESS_TOKEN": settings.gitea_token,
            "GITEA_HOST": settings.gitea_url,
            "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
        },
    )


def create_gitea_mcp_tools():
    """Create an MCPTools instance for the Gitea MCP server.

    Returns None if Gitea is disabled or not configured (no token).
    The returned MCPTools is lazy — Agno connects it on first ``arun()``.

    Uses ``server_params`` instead of ``command`` to bypass Agno's
    executable whitelist (gitea-mcp is a Go binary not in the list).
    """
    if not settings.gitea_enabled or not settings.gitea_token:
        logger.debug("Gitea MCP: disabled or no token configured")
        return None

    try:
        from agno.tools.mcp import MCPTools

        tools = MCPTools(
            server_params=_gitea_server_params(),
            include_tools=[
                "issue_write",
                "issue_read",
                "list_issues",
                "pull_request_write",
                "pull_request_read",
                "list_pull_requests",
                "list_branches",
                "list_commits",
            ],
            timeout_seconds=settings.mcp_timeout,
        )
        logger.info("Gitea MCP tools created (lazy connect)")
        return tools
    except Exception as exc:
        logger.warning("Failed to create Gitea MCP tools: %s", exc)
        return None


def create_filesystem_mcp_tools():
    """Create an MCPTools instance for the filesystem MCP server.

    Returns None if the command is not configured.
    Scoped to the project repo_root directory.

    Uses ``server_params`` for consistency (npx is whitelisted by Agno
    but server_params is the more robust approach).
    """
    try:
        from agno.tools.mcp import MCPTools
        from mcp.client.stdio import StdioServerParameters

        # Parse the base command, then append repo_root as an extra arg
        exe, args = _parse_command(settings.mcp_filesystem_command)
        args.append(str(settings.repo_root))

        params = StdioServerParameters(
            command=exe,
            args=args,
            env={
                "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
            },
        )

        tools = MCPTools(
            server_params=params,
            include_tools=[
                "read_file",
                "write_file",
                "list_directory",
                "search_files",
                "get_file_info",
                "directory_tree",
            ],
            timeout_seconds=settings.mcp_timeout,
        )
        logger.info("Filesystem MCP tools created (lazy connect)")
        return tools
    except Exception as exc:
        logger.warning("Failed to create filesystem MCP tools: %s", exc)
        return None


def _bridge_to_work_order(title: str, body: str, category: str) -> None:
    """Create a local work order so the dashboard tracks the issue."""
    try:
        db_path = Path(settings.repo_root) / "data" / "work_orders.db"
        db_path.parent.mkdir(parents=True, exist_ok=True)
        with closing(sqlite3.connect(str(db_path))) as conn:
            conn.execute(
                """CREATE TABLE IF NOT EXISTS work_orders (
                    id TEXT PRIMARY KEY,
                    title TEXT NOT NULL,
                    description TEXT DEFAULT '',
                    priority TEXT DEFAULT 'medium',
                    category TEXT DEFAULT 'suggestion',
                    submitter TEXT DEFAULT 'dashboard',
                    related_files TEXT DEFAULT '',
                    status TEXT DEFAULT 'submitted',
                    result TEXT DEFAULT '',
                    rejection_reason TEXT DEFAULT '',
                    created_at TEXT DEFAULT (datetime('now')),
                    completed_at TEXT
                )"""
            )
            conn.execute(
                "INSERT INTO work_orders (id, title, description, category, submitter, created_at) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (
                    str(uuid.uuid4()),
                    title,
                    body,
                    category,
                    "timmy-thinking",
                    datetime.utcnow().isoformat(),
                ),
            )
            conn.commit()
    except Exception as exc:
        logger.debug("Work order bridge failed: %s", exc)


async def create_gitea_issue_via_mcp(title: str, body: str = "", labels: str = "") -> str:
    """File a Gitea issue via the MCP server (standalone, no LLM loop).

    Used by the thinking engine's ``_maybe_file_issues()`` post-hook.
    Manages its own MCPTools session with lazy connect + graceful failure.

    Uses ``tools.session.call_tool()`` for direct MCP invocation — the
    ``MCPTools`` wrapper itself does not expose ``call_tool()``.

    Args:
        title: Issue title.
        body: Issue body (markdown).
        labels: Comma-separated label names.

    Returns:
        Confirmation string or error explanation.
    """
    if not settings.gitea_enabled or not settings.gitea_token:
        return "Gitea integration is not configured."

    try:
        from agno.tools.mcp import MCPTools

        global _issue_session

        if _issue_session is None:
            _issue_session = MCPTools(
                server_params=_gitea_server_params(),
                timeout_seconds=settings.mcp_timeout,
            )

        # Ensure connected
        if not getattr(_issue_session, "_connected", False):
            await _issue_session.connect()
            _issue_session._connected = True

        # Append auto-filing signature
        full_body = body
        if full_body:
            full_body += "\n\n"
        full_body += "---\n*Auto-filed by Timmy's thinking engine*"

        # Parse owner/repo from settings
        owner, repo = settings.gitea_repo.split("/", 1)

        # Build tool arguments — gitea-mcp uses issue_write with method="create"
        args = {
            "method": "create",
            "owner": owner,
            "repo": repo,
            "title": title,
            "body": full_body,
        }

        # Call via the underlying MCP session (MCPTools doesn't expose call_tool)
        result = await _issue_session.session.call_tool("issue_write", arguments=args)

        # Bridge to local work order
        label_list = [tag.strip() for tag in labels.split(",") if tag.strip()] if labels else []
        category = "bug" if "bug" in label_list else "suggestion"
        _bridge_to_work_order(title, body, category)

        logger.info("Created Gitea issue via MCP: %s", title[:60])
        return f"Created issue: {title}\n{result}"

    except Exception as exc:
        logger.warning("MCP issue creation failed: %s", exc)
        return f"Failed to create issue via MCP: {exc}"


def _draw_background(draw: ImageDraw.ImageDraw, size: int) -> None:
    """Draw radial gradient background with concentric circles."""
    for i in range(size // 2, 0, -4):
        g = int(25 + (i / (size // 2)) * 30)
        draw.ellipse(
            [size // 2 - i, size // 2 - i, size // 2 + i, size // 2 + i],
            fill=(10, g, 20),
        )


def _draw_wizard(draw: ImageDraw.ImageDraw) -> None:
    """Draw wizard hat, face, eyes, smile, monogram, and robe."""
    hat_color = (100, 50, 160)  # purple
    hat_outline = (180, 130, 255)
    gold = (220, 190, 50)
    pupil = (30, 30, 60)

    # Hat + brim
    draw.polygon([(256, 40), (160, 220), (352, 220)], fill=hat_color, outline=hat_outline)
    draw.ellipse([140, 200, 372, 250], fill=hat_color, outline=hat_outline)

    # Face
    draw.ellipse([190, 220, 322, 370], fill=(60, 180, 100), outline=(80, 220, 120))

    # Eyes (whites + pupils)
    draw.ellipse([220, 275, 248, 310], fill=(255, 255, 255))
    draw.ellipse([264, 275, 292, 310], fill=(255, 255, 255))
    draw.ellipse([228, 285, 242, 300], fill=pupil)
    draw.ellipse([272, 285, 286, 300], fill=pupil)

    # Smile
    draw.arc([225, 300, 287, 355], start=10, end=170, fill=pupil, width=3)

    # "T" monogram on hat
    draw.text((243, 100), "T", fill=gold)

    # Robe
    draw.polygon(
        [(180, 370), (140, 500), (372, 500), (332, 370)],
        fill=(40, 100, 70),
        outline=(60, 160, 100),
    )


def _draw_stars(draw: ImageDraw.ImageDraw) -> None:
    """Draw decorative gold stars around the wizard hat."""
    gold = (220, 190, 50)
    for sx, sy in [(120, 100), (380, 120), (100, 300), (400, 280), (256, 10)]:
        r = 8
        draw.polygon(
            [
                (sx, sy - r),
                (sx + r // 3, sy - r // 3),
                (sx + r, sy),
                (sx + r // 3, sy + r // 3),
                (sx, sy + r),
                (sx - r // 3, sy + r // 3),
                (sx - r, sy),
                (sx - r // 3, sy - r // 3),
            ],
            fill=gold,
        )


def _generate_avatar_image() -> bytes:
    """Generate a Timmy-themed avatar image using Pillow.

    Creates a 512x512 wizard-themed avatar with emerald/purple/gold palette.
    Returns raw PNG bytes. Falls back to a minimal solid-color image if
    Pillow drawing primitives fail.
    """
    import io

    from PIL import Image, ImageDraw

    size = 512
    img = Image.new("RGB", (size, size), (15, 25, 20))
    draw = ImageDraw.Draw(img)

    _draw_background(draw, size)
    _draw_wizard(draw)
    _draw_stars(draw)

    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return buf.getvalue()


async def update_gitea_avatar() -> str:
    """Generate and upload a unique avatar to Timmy's Gitea profile.

    Creates a wizard-themed avatar image using Pillow drawing primitives,
    base64-encodes it, and POSTs to the Gitea user avatar API endpoint.

    Returns:
        Success or failure message string.
    """
    if not settings.gitea_enabled or not settings.gitea_token:
        return "Gitea integration is not configured (no token or disabled)."

    try:
        from PIL import Image  # noqa: F401 — availability check
    except ImportError:
        return "Pillow is not installed — cannot generate avatar image."

    try:
        import base64

        # Step 1: Generate the avatar image
        png_bytes = _generate_avatar_image()
        logger.info("Generated avatar image (%d bytes)", len(png_bytes))

        # Step 2: Base64-encode (raw, no data URI prefix)
        b64_image = base64.b64encode(png_bytes).decode("ascii")

        # Step 3: POST to Gitea
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.post(
                f"{settings.gitea_url}/api/v1/user/avatar",
                headers={
                    "Authorization": f"token {settings.gitea_token}",
                    "Content-Type": "application/json",
                },
                json={"image": b64_image},
            )

        # Gitea returns empty body on success (204 or 200)
        if resp.status_code in (200, 204):
            logger.info("Gitea avatar updated successfully")
            return "Avatar updated successfully on Gitea."

        logger.warning("Gitea avatar update failed: %s %s", resp.status_code, resp.text[:200])
        return f"Gitea avatar update failed (HTTP {resp.status_code}): {resp.text[:200]}"

    except (httpx.ConnectError, httpx.ReadError, ConnectionError) as exc:
        logger.warning("Gitea connection failed during avatar update: %s", exc)
        return f"Could not connect to Gitea: {exc}"
    except Exception as exc:
        logger.error("Avatar update failed: %s", exc)
        return f"Avatar update failed: {exc}"


async def close_mcp_sessions() -> None:
    """Close any open MCP sessions. Called during app shutdown."""
    global _issue_session
    if _issue_session is not None:
        try:
            await _issue_session.close()
        except Exception as exc:
            logger.debug("MCP session close error: %s", exc)
        _issue_session = None