corvid/mcp_server.py

#!/usr/bin/env python3
"""
Corvid MCP server — homelab agent tools.

Provides ticket management and homelab monitoring tools for Claude Code agents.

Configure in ~/.claude/mcp.json:
  "homelab-tickets": {
    "command": "python3",
    "args": ["/home/caoimhinr/Projects/corvid/mcp_server.py"],
    "env": {
      "HOMEDASH_URL":   "https://homedash.welvaert.org",
      "HOMEDASH_API_KEY": "<TICKETS_API_KEY value>"
    }
  }
"""

import os

import httpx
from mcp.server.fastmcp import FastMCP

HOMEDASH_URL = os.environ.get("HOMEDASH_URL", "http://localhost:8080").rstrip("/")
API_KEY      = os.environ.get("HOMEDASH_API_KEY", "")

mcp = FastMCP("homelab-tickets")

_HEADERS = {"X-Api-Key": API_KEY, "Content-Type": "application/json"}

# Repo root used in generated startup scripts (cd here before running claude)
_REPO_PATH = os.environ.get("CORVID_REPO_PATH",
                             os.path.expanduser("~/Projects/corvid"))

# MCP tool names always included so the agent can read and update its own ticket
_BASE_TOOLS = [
    "Read",
    "mcp__homelab-tickets__get_ticket",
    "mcp__homelab-tickets__update_ticket",
    "mcp__homelab-tickets__add_attachment",
    "mcp__homelab-tickets__list_attachments",
]

_HA_KEYWORDS = (
    "home assistant", " ha ", "climate", "temperature", "thermostat",
    "light", "sensor", "entity", "automation", "scene",
)
_WEB_KEYWORDS = (
    "research", "investigate", "look up", "documentation", "readme",
    " api ", "spec ", "rfc",
)


def _generate_startup_script(ticket_id: int, human_id: str, title: str,
                              description: str | None, ticket_type: str) -> str:
    text  = f"{title} {description or ''}".lower()
    tools = list(_BASE_TOOLS)
    if ticket_type in ("feature", "bug", "chore"):
        tools += ["Edit", "Write", "Bash"]
    if any(kw in text for kw in _WEB_KEYWORDS):
        tools += ["WebSearch", "WebFetch"]
    if any(kw in text for kw in _HA_KEYWORDS):
        tools += [
            "mcp__homeassistant__GetLiveContext",
            "mcp__homeassistant__HassTurnOn",
            "mcp__homeassistant__HassTurnOff",
            "mcp__homeassistant__HassLightSet",
            "mcp__homeassistant__HassClimateSetTemperature",
        ]
    tools_str = ",".join(tools)
    prompt = (
        f"complete ticket {human_id}: {title}. "
        f"Start by calling mcp__homelab-tickets__get_ticket({ticket_id}) to read the full "
        f"ticket details, then complete the work and mark the ticket as completed."
    )
    return (
        f"#!/usr/bin/env bash\n"
        f"# Startup script for {human_id}: {title}\n"
        f"cd {_REPO_PATH}\n"
        f'claude "{prompt}" \\\n'
        f'  --allowedTools "{tools_str}"\n'
    )


def _client() -> httpx.Client:
    return httpx.Client(base_url=HOMEDASH_URL, headers=_HEADERS, timeout=10.0)


def _check(r: httpx.Response) -> dict | list:
    if not r.is_success:
        raise RuntimeError(f"API error {r.status_code}: {r.text}")
    return r.json()


# ── Environment tools ─────────────────────────────────────────────────────────

@mcp.tool()
def list_environments() -> list[dict]:
    """
    List all ticket environments (prefix + name). Each environment scopes a
    set of tickets with its own ID prefix (e.g. HOME-001, PROJ-001).
    """
    with _client() as c:
        return _check(c.get("/api/environments"))


@mcp.tool()
def create_environment(prefix: str, name: str, description: str = "") -> dict:
    """
    Create a new ticket environment.

    Args:
        prefix:      Short uppercase identifier, e.g. PROJ or LAB (max 16 chars).
        name:        Human-readable name, e.g. "My Project".
        description: Optional longer description.
    """
    with _client() as c:
        return _check(c.post("/api/environments", json={
            "prefix":      prefix.upper(),
            "name":        name,
            "description": description or None,
        }))


# ── Ticket tools ──────────────────────────────────────────────────────────────

@mcp.tool()
def list_tickets(status: str = "", type: str = "", q: str = "",
                 environment_id: int = 0) -> list[dict]:
    """
    List homelab tickets. Results include attachment_count so you know
    which tickets have supporting documents without fetching them.

    Args:
        status:         Filter by status — open, ongoing, completed, abandoned.
                        Leave empty to list all statuses.
        type:           Filter by type — feature, bug, chore, project.
                        Leave empty to list all types.
        q:              Full-text search query matched against ticket title and
                        description (case-insensitive). Leave empty to skip.
        environment_id: Filter by environment ID. Pass 0 (default) for all environments.
    """
    with _client() as c:
        params = {}
        if status:
            params["status"] = status
        if type:
            params["type"] = type
        if q:
            params["q"] = q
        if environment_id:
            params["environment_id"] = environment_id
        return _check(c.get("/api/tickets", params=params))


@mcp.tool()
def get_ticket(ticket_id: int) -> dict:
    """
    Get a single ticket by its numeric ID, including attachment_count.
    Use list_attachments to fetch the actual attachment content.

    Args:
        ticket_id: The integer primary key (e.g. 1 for HOME-001).
    """
    with _client() as c:
        return _check(c.get(f"/api/tickets/{ticket_id}"))


@mcp.tool()
def create_ticket(
    title:          str,
    description:    str = "",
    status:         str = "open",
    type:           str = "feature",
    user:           str = "claude",
    parent_id:      int = 0,
    effort:         int = 0,
    environment_id: int = 0,
) -> dict:
    """
    Create a new homelab ticket. For longer supporting documents
    (proposals, reports, specs) prefer add_attachment over stuffing
    everything into description.

    Args:
        title:          Short summary of the work item.
        description:    Markdown body with context, links, acceptance criteria.
        status:         open (default), ongoing, completed, abandoned.
        type:           feature (default), bug, chore, project.
        user:           Who is submitting — kevin or claude (default: claude).
        parent_id:      ID of the parent ticket (omit or pass 0 for no parent).
        effort:         Estimated effort rating 1–10 (omit or pass 0 to leave unset).
        environment_id: Environment to file the ticket in. Pass 0 to use the
                        default environment (HOME). Use list_environments() to
                        find available environments.
    """
    body: dict = {
        "title":       title,
        "description": description or None,
        "status":      status,
        "type":        type,
        "user":        user,
    }
    if parent_id:
        body["parent_id"] = parent_id
    if effort:
        body["effort"] = effort
    if environment_id:
        body["environment_id"] = environment_id
    with _client() as c:
        ticket = _check(c.post("/api/tickets", json=body))
        script = _generate_startup_script(
            ticket["id"], ticket["human_id"], title, description, type
        )
        ticket = _check(c.patch(f"/api/tickets/{ticket['id']}", json={"startup_script": script}))
        return ticket


@mcp.tool()
def delete_ticket(ticket_id: int) -> dict:
    """
    Permanently delete a ticket and all its attachments.

    Args:
        ticket_id: The integer ID of the ticket to delete.
    """
    with _client() as c:
        return _check(c.delete(f"/api/tickets/{ticket_id}"))


@mcp.tool()
def update_ticket(
    ticket_id:      int,
    title:          str = "",
    description:    str = "",
    status:         str = "",
    type:           str = "",
    user:           str = "",
    parent_id:      int = -1,
    effort:         int = -1,
    environment_id: int = -1,
) -> dict:
    """
    Update an existing homelab ticket. Only provided (non-empty) fields
    are changed; omit a field to leave it unchanged.

    Args:
        ticket_id:      The integer ID of the ticket to update.
        title:          New title.
        description:    New markdown description.
        status:         open, ongoing, completed, abandoned.
        type:           feature, bug, chore, project.
        user:           kevin or claude.
        parent_id:      Set parent ticket ID; pass 0 to clear the parent.
                        Omit (default -1) to leave unchanged.
        effort:         Effort rating 1–10; pass 0 to clear it.
                        Omit (default -1) to leave unchanged.
        environment_id: Move ticket to a different environment by ID.
                        Omit (default -1) to leave unchanged.
    """
    body = {}
    if title:       body["title"]       = title
    if description: body["description"] = description
    if status:      body["status"]      = status
    if type:        body["type"]        = type
    if user:        body["user"]        = user
    if parent_id >= 0:
        body["parent_id"] = parent_id if parent_id > 0 else None
    if effort >= 0:
        body["effort"] = effort if effort > 0 else None
    if environment_id >= 0:
        body["environment_id"] = environment_id if environment_id > 0 else None
    if not body:
        raise ValueError("Provide at least one field to update")
    with _client() as c:
        ticket = _check(c.patch(f"/api/tickets/{ticket_id}", json=body))
        if title or description or type:
            script = _generate_startup_script(
                ticket["id"], ticket["human_id"],
                ticket["title"], ticket["description"], ticket["type"],
            )
            ticket = _check(c.patch(f"/api/tickets/{ticket_id}", json={"startup_script": script}))
        return ticket


# ── Attachment tools ──────────────────────────────────────────────────────────

@mcp.tool()
def list_attachments(ticket_id: int) -> list[dict]:
    """
    List all markdown attachments on a ticket (title, created_by,
    created_at, and full content).

    Args:
        ticket_id: The integer ID of the parent ticket.
    """
    with _client() as c:
        return _check(c.get(f"/api/tickets/{ticket_id}/attachments"))


@mcp.tool()
def add_attachment(
    ticket_id:  int,
    title:      str,
    content:    str,
    created_by: str = "claude",
) -> dict:
    """
    Attach a markdown document to a ticket. Use this for implementation
    proposals, test plans, research notes, decision records, or any
    supporting document that is too long for the ticket description.

    Args:
        ticket_id:  The integer ID of the parent ticket.
        title:      Descriptive name, e.g. "Implementation Proposal v1".
        content:    Full markdown body of the document.
        created_by: kevin or claude (default: claude).
    """
    with _client() as c:
        return _check(c.post(f"/api/tickets/{ticket_id}/attachments", json={
            "title":      title,
            "content":    content,
            "created_by": created_by,
        }))


@mcp.tool()
def delete_attachment(ticket_id: int, attachment_id: int) -> dict:
    """
    Permanently delete an attachment from a ticket.

    Args:
        ticket_id:     The integer ID of the parent ticket.
        attachment_id: The integer ID of the attachment to delete.
    """
    with _client() as c:
        return _check(c.delete(f"/api/tickets/{ticket_id}/attachments/{attachment_id}"))


# ── System log tools ──────────────────────────────────────────────────────────

@mcp.tool()
def list_logs(
    level:    str  = "",
    source:   str  = "",
    resolved: bool = False,
) -> list[dict]:
    """
    List homelab system log entries (infrastructure alerts, camera issues, disk).

    Args:
        level:    Filter by level — ERR, WARN, INFO. Leave empty for all.
        source:   Filter by source — homelab, hassgrab, disk. Leave empty for all.
        resolved: Include resolved logs (default False = active only).
    """
    with _client() as c:
        params: dict = {"resolved": str(resolved).lower()}
        if level:
            params["level"] = level
        if source:
            params["source"] = source
        return _check(c.get("/api/logs", params=params))


if __name__ == "__main__":
    mcp.run(transport="stdio")