feat: initial corvid — standalone ticket service extracted from homelab-dashboard
Extracts the ticket/attachment system into a reusable pip-installable package. Centralises homelab MCP server (tickets, attachments, system logs) here. - corvid package: models, router (make_router factory), config, standalone main - Alembic migration chain for fresh standalone deployments - Docker Compose for standalone deployment (port 8090) - mcp_server.py: moved from homelab-dashboard, CORVID_REPO_PATH configurable Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
caoimhinr
commit
37b2c21720
27 changed files with 1072 additions and 0 deletions
320
mcp_server.py
Normal file
320
mcp_server.py
Normal file
|
|
@ -0,0 +1,320 @@
|
|||
#!/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()
|
||||
|
||||
|
||||
# ── Ticket tools ──────────────────────────────────────────────────────────────
|
||||
|
||||
@mcp.tool()
|
||||
def list_tickets(status: str = "", type: str = "", q: str = "") -> 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.
|
||||
"""
|
||||
with _client() as c:
|
||||
params = {}
|
||||
if status:
|
||||
params["status"] = status
|
||||
if type:
|
||||
params["type"] = type
|
||||
if q:
|
||||
params["q"] = q
|
||||
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,
|
||||
) -> 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).
|
||||
"""
|
||||
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
|
||||
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,
|
||||
) -> 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.
|
||||
"""
|
||||
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 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")
|
||||
Loading…
Add table
Add a link
Reference in a new issue