DevOpsDash/app/mcp_client.py

"""MCP proxy client — calls DevOpsMCP's worklog/standup tools over HTTP MCP protocol."""

from __future__ import annotations

import os
from typing import Any, Dict, Optional

import httpx

MCP_URL = os.environ.get("DEVOPS_MCP_URL", "http://localhost:8000")
_MCP_ENDPOINT = f"{MCP_URL}/mcp"


async def _call_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Call a DevOpsMCP tool via MCP JSON-RPC 2.0 over HTTP."""
    payload = {
        "jsonrpc": "2.0",
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments},
        "id": 1,
    }
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.post(_MCP_ENDPOINT, json=payload)
        resp.raise_for_status()
        data = resp.json()
        if "error" in data:
            raise RuntimeError(f"MCP error: {data['error']}")
        result = data.get("result", {})
        content = result.get("content", [])
        if content and isinstance(content[0], dict):
            import json
            text = content[0].get("text", "{}")
            try:
                return json.loads(text)
            except json.JSONDecodeError:
                return {"raw": text}
        return result


async def get_worklog(
    context: str = "egmont",
    days: int = 7,
    group_by: str = "repo",
    since_date: Optional[str] = None,
    until_date: Optional[str] = None,
) -> Dict[str, Any]:
    args: Dict[str, Any] = {"context": context, "days": days, "group_by": group_by}
    if since_date:
        args["since_date"] = since_date
    if until_date:
        args["until_date"] = until_date
    return await _call_tool("worklog", args)


async def get_standup(days: int = 2, context: str = "egmont") -> Dict[str, Any]:
    return await _call_tool("generate_standup", {"days": days, "context": context})


async def list_knowledge(category: Optional[str] = None, tag: Optional[str] = None) -> Dict[str, Any]:
    args: Dict[str, Any] = {}
    if category:
        args["category"] = category
    if tag:
        args["tag"] = tag
    return await _call_tool("list_knowledge", args)


async def get_knowledge(storage_filename: str) -> Dict[str, Any]:
    return await _call_tool("get_knowledge", {"storage_filename": storage_filename, "include_metadata": False})


async def list_howtos() -> Dict[str, Any]:
    return await _call_tool("list_howtos", {})


async def get_howto(filename: str) -> Dict[str, Any]:
    return await _call_tool("get_howto_content", {"filename": filename})


async def list_agents(domain: Optional[str] = None) -> Dict[str, Any]:
    args: Dict[str, Any] = {}
    if domain:
        args["domain"] = domain
    return await _call_tool("get_agent", args)


async def get_agent(name: str) -> Dict[str, Any]:
    return await _call_tool("get_agent", {"name": name})


async def list_skills(domain: Optional[str] = None) -> Dict[str, Any]:
    args: Dict[str, Any] = {}
    if domain:
        args["domain"] = domain
    return await _call_tool("get_skill", args)


async def get_skill(name: str) -> Dict[str, Any]:
    return await _call_tool("get_skill", {"name": name})