feat: add mcp server

api/mcp/__init__.py

@@ -0,0 +1,3 @@
+from api.mcp.server import mcp
+
+__all__ = ["mcp"]

api/mcp/auth.py

@@ -0,0 +1,25 @@
+from fastapi import HTTPException
+from fastmcp.server.dependencies import get_http_headers
+
+from api.db.models import UserModel
+from api.services.auth.depends import _handle_api_key_auth
+
+
+async def authenticate_mcp_request() -> UserModel:
+    """Resolve the authenticated Dograh user for an MCP tool invocation.
+
+    Accepts either `X-API-Key: <key>` or `Authorization: Bearer <key>`,
+    reusing the API-key flow from `api.services.auth.depends`.
+    """
+    headers = get_http_headers()
+    api_key = headers.get("x-api-key")
+    if not api_key:
+        auth = headers.get("authorization", "")
+        if auth.lower().startswith("bearer "):
+            api_key = auth.split(" ", 1)[1].strip()
+    if not api_key:
+        raise HTTPException(
+            status_code=401,
+            detail="Missing API key — send X-API-Key or Authorization: Bearer <key>",
+        )
+    return await _handle_api_key_auth(api_key)

api/mcp/server.py

@@ -0,0 +1,6 @@
+from fastmcp import FastMCP
+
+mcp = FastMCP("dograh")
+
+from api.mcp.tools import docs as _docs  # noqa: E402, F401
+from api.mcp.tools import workflows as _workflows  # noqa: E402, F401

api/mcp/tools/docs.py

@@ -0,0 +1,115 @@
+import re
+from functools import lru_cache
+from pathlib import Path
+
+from fastapi import HTTPException
+from rank_bm25 import BM25Okapi
+
+from api.mcp.server import mcp
+
+DOCS_ROOT = Path(__file__).resolve().parents[3] / "docs"
+
+_TOKEN_RE = re.compile(r"[A-Za-z0-9_]+")
+_FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL)
+_TITLE_RE = re.compile(r"^title:\s*['\"]?(.+?)['\"]?\s*$", re.MULTILINE)
+_H1_RE = re.compile(r"^#\s+(.+?)\s*$", re.MULTILINE)
+
+
+def _tokenize(text: str) -> list[str]:
+    return [t.lower() for t in _TOKEN_RE.findall(text)]
+
+
+def _extract_title(path: Path, body: str) -> str:
+    fm_match = _FRONTMATTER_RE.match(body)
+    if fm_match:
+        title_match = _TITLE_RE.search(fm_match.group(1))
+        if title_match:
+            return title_match.group(1).strip()
+    h1_match = _H1_RE.search(body)
+    if h1_match:
+        return h1_match.group(1).strip()
+    return path.stem.replace("-", " ").title()
+
+
+def _strip_frontmatter(body: str) -> str:
+    return _FRONTMATTER_RE.sub("", body, count=1)
+
+
+@lru_cache(maxsize=1)
+def _load_index() -> tuple[list[dict], BM25Okapi]:
+    """Read every docs/**/*.mdx file once and build a BM25 index.
+
+    Cached for the process lifetime — docs rarely change between restarts.
+    """
+    docs: list[dict] = []
+    corpus: list[list[str]] = []
+
+    for path in sorted(DOCS_ROOT.rglob("*.mdx")):
+        body = path.read_text(encoding="utf-8")
+        rel = path.relative_to(DOCS_ROOT).as_posix()
+        title = _extract_title(path, body)
+        content = _strip_frontmatter(body)
+        docs.append({"path": rel, "title": title, "content": content})
+        corpus.append(_tokenize(f"{title} {content}"))
+
+    return docs, BM25Okapi(corpus)
+
+
+def _snippet(content: str, query_tokens: list[str], width: int = 240) -> str:
+    lowered = content.lower()
+    for tok in query_tokens:
+        idx = lowered.find(tok)
+        if idx >= 0:
+            start = max(0, idx - width // 2)
+            end = min(len(content), start + width)
+            return ("…" if start > 0 else "") + content[start:end].strip() + (
+                "…" if end < len(content) else ""
+            )
+    return content[:width].strip() + ("…" if len(content) > width else "")
+
+
+@mcp.tool
+async def search_dograh_docs(query: str, limit: int = 5) -> list[dict]:
+    """Search Dograh's product documentation.
+
+    Returns the top matches as {path, title, snippet}. Pass the returned
+    `path` to `fetch_dograh_doc` to read the full page. Use this first
+    when you need to learn how a Dograh feature works before building
+    against it.
+    """
+    docs, bm25 = _load_index()
+    tokens = _tokenize(query)
+    if not tokens:
+        return []
+
+    scores = bm25.get_scores(tokens)
+    ranked = sorted(
+        zip(scores, docs), key=lambda pair: pair[0], reverse=True
+    )[:limit]
+
+    return [
+        {
+            "path": doc["path"],
+            "title": doc["title"],
+            "snippet": _snippet(doc["content"], tokens),
+            "score": round(float(score), 3),
+        }
+        for score, doc in ranked
+        if score > 0
+    ]
+
+
+@mcp.tool
+async def fetch_dograh_doc(path: str) -> dict:
+    """Fetch the full content of a Dograh docs page by its path
+    (e.g. `core-concepts/workflows.mdx`), as returned by `search_dograh_docs`.
+    """
+    docs, _ = _load_index()
+    for doc in docs:
+        if doc["path"] == path:
+            return {
+                "path": doc["path"],
+                "title": doc["title"],
+                "content": doc["content"],
+            }
+    raise HTTPException(status_code=404, detail=f"Doc not found: {path}")

api/mcp/tools/workflows.py

@@ -0,0 +1,47 @@
+from fastapi import HTTPException
+
+from api.db import db_client
+from api.mcp.auth import authenticate_mcp_request
+from api.mcp.server import mcp
+
+
+@mcp.tool
+async def list_workflows(status: str | None = None) -> list[dict]:
+    """List agents (workflows) in the caller's organization.
+
+    Returns id, name, status, and created_at for each agent. Use
+    `get_workflow` to fetch a single agent's full definition. Pass
+    `status="active"` or `status="archived"` to filter.
+    """
+    user = await authenticate_mcp_request()
+    workflows = await db_client.get_all_workflows_for_listing(
+        organization_id=user.selected_organization_id,
+        status=status,
+    )
+    return [
+        {
+            "id": w.id,
+            "name": w.name,
+            "status": w.status,
+            "created_at": w.created_at.isoformat() if w.created_at else None,
+        }
+        for w in workflows
+    ]
+
+
+@mcp.tool
+async def get_workflow(workflow_id: int) -> dict:
+    """Fetch a single agent by id, including its current published definition."""
+    user = await authenticate_mcp_request()
+    workflow = await db_client.get_workflow_by_id(workflow_id)
+    if not workflow or workflow.organization_id != user.selected_organization_id:
+        raise HTTPException(status_code=404, detail=f"Workflow {workflow_id} not found")
+
+    current = workflow.current_definition
+    return {
+        "id": workflow.id,
+        "name": workflow.name,
+        "status": workflow.status,
+        "definition": current.workflow_json if current else None,
+        "version_number": current.version_number if current else None,
+    }