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
import os
import shutil
import sqlite3
import uuid
from datetime import datetime
from pathlib import Path

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)
        conn = sqlite3.connect(str(db_path))
        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()
        conn.close()
    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}"


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