Implement alert_mcp_server with Gradio and httpx

pyproject.toml

@@ -1,19 +1,19 @@
 [project]
-name = "alert_mcp"
+name = "alert_mcp_server"
 version = "0.1.0"
-description = "MCP server for CredentialWatch alerts"
+description = "Gradio MCP server for CredentialWatch alerts"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    "fastapi",
-    "uvicorn",
-    "sqlalchemy",
+    "gradio[mcp]",
+    "httpx",
     "pydantic",
-    "mcp",  # Assuming a generic mcp package or just standard http tools
+    "python-dotenv",
+    "mcp"
 ]
 
 [tool.uv]
 dev-dependencies = [
     "pytest",
-    "httpx",
+    "pytest-asyncio"
 ]

src/alert_mcp_server/config.py

@@ -0,0 +1,6 @@
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+ALERT_API_BASE_URL = os.getenv("ALERT_API_BASE_URL", "http://localhost:8000")

src/alert_mcp_server/main.py

@@ -0,0 +1,80 @@
+import gradio as gr
+from .tools import log_alert, get_open_alerts, mark_alert_resolved, summarize_alerts
+
+# Define the Gradio interface
+# We can use a TabbedInterface to organize the tools for the UI,
+# which also registers them as MCP tools.
+
+def create_demo():
+    with gr.Blocks(title="CredentialWatch Alert MCP") as demo:
+        gr.Markdown("# CredentialWatch Alert MCP Server")
+        gr.Markdown("This server exposes tools for managing alerts via MCP. It also provides a simple UI.")
+
+        with gr.Tab("Log Alert"):
+            gr.Markdown("## Log a new alert")
+            with gr.Row():
+                t1_provider_id = gr.Number(label="Provider ID", value=1, precision=0)
+                t1_credential_id = gr.Number(label="Credential ID (Optional)", value=None, precision=0)
+
+            with gr.Row():
+                t1_severity = gr.Dropdown(choices=["info", "warning", "critical"], label="Severity", value="info")
+                t1_window_days = gr.Number(label="Window Days", value=30, precision=0)
+
+            t1_message = gr.Textbox(label="Message")
+            t1_channel = gr.Textbox(label="Channel", value="ui")
+
+            t1_output = gr.JSON(label="Response")
+            t1_btn = gr.Button("Log Alert")
+
+            t1_btn.click(
+                fn=log_alert,
+                inputs=[t1_provider_id, t1_severity, t1_window_days, t1_message, t1_credential_id, t1_channel],
+                outputs=t1_output
+            )
+
+        with gr.Tab("Get Open Alerts"):
+            gr.Markdown("## List open alerts")
+            t2_provider_id = gr.Number(label="Provider ID (Optional)", value=None, precision=0)
+            t2_severity = gr.Dropdown(choices=["info", "warning", "critical", None], label="Severity (Optional)", value=None)
+
+            t2_output = gr.JSON(label="Open Alerts")
+            t2_btn = gr.Button("Get Alerts")
+
+            t2_btn.click(
+                fn=get_open_alerts,
+                inputs=[t2_provider_id, t2_severity],
+                outputs=t2_output
+            )
+
+        with gr.Tab("Resolve Alert"):
+            gr.Markdown("## Mark alert as resolved")
+            t3_alert_id = gr.Number(label="Alert ID", precision=0)
+            t3_note = gr.Textbox(label="Resolution Note")
+
+            t3_output = gr.JSON(label="Updated Alert")
+            t3_btn = gr.Button("Resolve")
+
+            t3_btn.click(
+                fn=mark_alert_resolved,
+                inputs=[t3_alert_id, t3_note],
+                outputs=t3_output
+            )
+
+        with gr.Tab("Summarize Alerts"):
+            gr.Markdown("## Summary of alerts")
+            t4_window = gr.Number(label="Window Days (Optional)", value=None, precision=0)
+
+            t4_output = gr.JSON(label="Summary")
+            t4_btn = gr.Button("Summarize")
+
+            t4_btn.click(
+                fn=summarize_alerts,
+                inputs=[t4_window],
+                outputs=t4_output
+            )
+
+    return demo
+
+if __name__ == "__main__":
+    demo = create_demo()
+    demo.launch(mcp_server=True)

src/alert_mcp_server/schemas.py

@@ -0,0 +1,29 @@
+from typing import Optional, Literal
+from pydantic import BaseModel, Field
+from datetime import datetime
+
+class AlertCreate(BaseModel):
+    provider_id: int
+    credential_id: Optional[int] = None
+    severity: Literal["info", "warning", "critical"]
+    window_days: int
+    message: str
+    channel: Optional[str] = "ui"
+
+class AlertRead(BaseModel):
+    id: int
+    provider_id: int
+    credential_id: Optional[int] = None
+    severity: str
+    window_days: int
+    message: str
+    channel: str
+    created_at: datetime
+    resolved_at: Optional[datetime] = None
+    resolution_note: Optional[str] = None
+
+class AlertResolution(BaseModel):
+    resolution_note: Optional[str] = None
+
+class AlertSummary(BaseModel):
+    severity_counts: dict[str, int]

src/alert_mcp_server/tests/test_alert_mcp_server.py

@@ -0,0 +1,104 @@
+
+import pytest
+import httpx
+from unittest.mock import AsyncMock, MagicMock, patch
+from src.alert_mcp_server.tools import log_alert, get_open_alerts, mark_alert_resolved, summarize_alerts
+
+@pytest.mark.asyncio
+async def test_log_alert():
+    mock_response = {
+        "id": 1,
+        "provider_id": 1,
+        "severity": "info",
+        "message": "Test alert",
+        "created_at": "2023-10-27T10:00:00"
+    }
+
+    # We use new_callable=AsyncMock for the client.post method because it is awaited.
+    # However, the return value (the response object) should be a synchronous Mock (MagicMock)
+    # because methods like .json() and .raise_for_status() are synchronous on the Response object.
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
+        mock_response_obj = MagicMock()
+        mock_response_obj.status_code = 200
+        mock_response_obj.json.return_value = mock_response
+        mock_response_obj.raise_for_status.return_value = None
+
+        mock_post.return_value = mock_response_obj
+
+        result = await log_alert(
+            provider_id=1,
+            severity="info",
+            window_days=30,
+            message="Test alert"
+        )
+
+        assert result == mock_response
+        mock_post.assert_called_once()
+        args, kwargs = mock_post.call_args
+        assert kwargs["json"]["provider_id"] == 1
+        assert kwargs["json"]["severity"] == "info"
+
+@pytest.mark.asyncio
+async def test_get_open_alerts():
+    mock_response = [
+        {
+            "id": 1,
+            "provider_id": 1,
+            "severity": "critical",
+            "message": "Critical alert",
+            "created_at": "2023-10-27T10:00:00"
+        }
+    ]
+
+    with patch("httpx.AsyncClient.get", new_callable=AsyncMock) as mock_get:
+        mock_response_obj = MagicMock()
+        mock_response_obj.status_code = 200
+        mock_response_obj.json.return_value = mock_response
+        mock_response_obj.raise_for_status.return_value = None
+
+        mock_get.return_value = mock_response_obj
+
+        result = await get_open_alerts(provider_id=1)
+
+        assert result == mock_response
+        mock_get.assert_called_once()
+        args, kwargs = mock_get.call_args
+        assert kwargs["params"]["provider_id"] == 1
+
+@pytest.mark.asyncio
+async def test_mark_alert_resolved():
+    mock_response = {
+        "id": 1,
+        "resolved_at": "2023-10-27T12:00:00",
+        "resolution_note": "Fixed"
+    }
+
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
+        mock_response_obj = MagicMock()
+        mock_response_obj.status_code = 200
+        mock_response_obj.json.return_value = mock_response
+        mock_response_obj.raise_for_status.return_value = None
+
+        mock_post.return_value = mock_response_obj
+
+        result = await mark_alert_resolved(alert_id=1, resolution_note="Fixed")
+
+        assert result == mock_response
+        mock_post.assert_called_once()
+
+@pytest.mark.asyncio
+async def test_summarize_alerts():
+    mock_response = {"info": 5, "warning": 2, "critical": 1}
+
+    with patch("httpx.AsyncClient.post", new_callable=AsyncMock) as mock_post:
+        mock_response_obj = MagicMock()
+        mock_response_obj.status_code = 200
+        mock_response_obj.json.return_value = mock_response
+        mock_response_obj.raise_for_status.return_value = None
+
+        mock_post.return_value = mock_response_obj
+
+        result = await summarize_alerts(window_days=7)
+
+        assert result == mock_response
+        mock_post.assert_called_once()

src/alert_mcp_server/tools.py

@@ -0,0 +1,132 @@
+import httpx
+from typing import List, Optional, Dict, Any
+from .config import ALERT_API_BASE_URL
+from .schemas import AlertCreate, AlertRead, AlertResolution
+
+# We'll use a synchronous client for simplicity in Gradio functions,
+# but Gradio supports async too. Let's use synchronous for now as it's often easier with Gradio tools.
+# Actually, httpx is great for async, but standard requests is often used.
+# Let's stick to httpx but use it synchronously or asynchronously depending on how we define the tools.
+# Gradio tools can be async def.
+
+async def log_alert(
+    provider_id: int,
+    severity: str,
+    window_days: int,
+    message: str,
+    credential_id: Optional[int] = None,
+    channel: Optional[str] = "ui"
+) -> Dict[str, Any]:
+    """
+    Log a new alert in the system.
+
+    Args:
+        provider_id: The ID of the provider.
+        severity: One of "info", "warning", "critical".
+        window_days: The window in days.
+        message: The alert message.
+        credential_id: Optional credential ID.
+        channel: Notification channel (default: "ui").
+    """
+    # Validation logic is handled by Pydantic model implicitly if we use it,
+    # but here we are constructing the payload.
+    # Basic validation for severity
+    if severity not in ["info", "warning", "critical"]:
+        raise ValueError("Severity must be one of: info, warning, critical")
+
+    payload = {
+        "provider_id": provider_id,
+        "credential_id": credential_id,
+        "severity": severity,
+        "window_days": window_days,
+        "message": message,
+        "channel": channel
+    }
+
+    async with httpx.AsyncClient(base_url=ALERT_API_BASE_URL) as client:
+        try:
+            response = await client.post("/alerts", json=payload)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            return {"error": f"HTTP error occurred: {e.response.status_code} - {e.response.text}"}
+        except Exception as e:
+            return {"error": f"An error occurred: {str(e)}"}
+
+async def get_open_alerts(
+    provider_id: Optional[int] = None,
+    severity: Optional[str] = None
+) -> List[Dict[str, Any]]:
+    """
+    Get a list of open (unresolved) alerts.
+
+    Args:
+        provider_id: Optional filter by provider ID.
+        severity: Optional filter by severity.
+    """
+    params = {}
+    if provider_id is not None:
+        params["provider_id"] = provider_id
+    if severity is not None:
+        params["severity"] = severity
+
+    # Using POST /alerts/open with filters as per instructions, or GET if API supports params.
+    # Instruction says: "GET or POST to /alerts/open with filters."
+    # I'll try POST as it's more flexible for filters usually, or standard GET with query params.
+    # Let's assume POST for filters body if complex, or GET with query params.
+    # I will assume GET with query params first as it's standard for 'get', but the instruction says "GET or POST".
+    # I will implement as GET with query params for now.
+
+    async with httpx.AsyncClient(base_url=ALERT_API_BASE_URL) as client:
+        try:
+            response = await client.get("/alerts/open", params=params)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            return [{"error": f"HTTP error: {e.response.status_code}"}]
+        except Exception as e:
+            return [{"error": str(e)}]
+
+async def mark_alert_resolved(
+    alert_id: int,
+    resolution_note: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Mark an alert as resolved.
+
+    Args:
+        alert_id: The ID of the alert to resolve.
+        resolution_note: A note explaining the resolution.
+    """
+    payload = {"resolution_note": resolution_note}
+
+    async with httpx.AsyncClient(base_url=ALERT_API_BASE_URL) as client:
+        try:
+            response = await client.post(f"/alerts/{alert_id}/resolve", json=payload)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            return {"error": f"HTTP error: {e.response.status_code}"}
+        except Exception as e:
+            return {"error": str(e)}
+
+async def summarize_alerts(window_days: Optional[int] = None) -> Dict[str, int]:
+    """
+    Get a summary of alerts (counts by severity).
+
+    Args:
+        window_days: Optional window in days to summarize over.
+    """
+    payload = {}
+    if window_days is not None:
+        payload["window_days"] = window_days
+
+    async with httpx.AsyncClient(base_url=ALERT_API_BASE_URL) as client:
+        try:
+            response = await client.post("/alerts/summary", json=payload)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            return {"error": f"HTTP error: {e.response.status_code}"}
+        except Exception as e:
+            return {"error": str(e)}