verifywise/AIGateway/src/routers/mcp_audit.py at 435a111c271eca227de430df05ab98535cd2b588 · verifywise-ai/verifywise · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from typing import Optional

from fastapi import APIRouter, Request, status

from crud.mcp_audit import (
    delete_expired_audit_logs,
    get_audit_logs,
    get_audit_stats,
    get_audit_stats_by_agent,
    get_audit_stats_by_tool,
)
from middlewares.auth import verify_internal_key
from utils.auth import get_org_id

router = APIRouter(prefix="/mcp/audit", tags=["mcp-audit"])


# ---------------------------------------------------------------------------
# GET /mcp/audit/logs
# ---------------------------------------------------------------------------

@router.get("/logs", status_code=status.HTTP_200_OK)
async def list_audit_logs(
    request: Request,
    limit: int = 50,
    offset: int = 0,
    agent_key_id: Optional[int] = None,
    tool_name: Optional[str] = None,
    result_status: Optional[str] = None,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
):
    verify_internal_key(request)
    org_id = get_org_id(request)

    filters: dict = {}
    if agent_key_id is not None:
        filters["agent_key_id"] = agent_key_id
    if tool_name is not None:
        filters["tool_name"] = tool_name
    if result_status is not None:
        filters["result_status"] = result_status
    if start_date is not None:
        filters["start_date"] = start_date
    if end_date is not None:
        filters["end_date"] = end_date

    result = await get_audit_logs(org_id, limit=limit, offset=offset, filters=filters)
    return {
        "status": "success",
        "data": result["data"],
        "total": result["total"],
        "limit": result["limit"],
        "offset": result["offset"],
    }


# ---------------------------------------------------------------------------
# GET /mcp/audit/stats
# ---------------------------------------------------------------------------

@router.get("/stats", status_code=status.HTTP_200_OK)
async def audit_stats(request: Request, days: int = 7):
    verify_internal_key(request)
    org_id = get_org_id(request)

    stats = await get_audit_stats(org_id, days=days)
    return {"status": "success", "data": stats}


# ---------------------------------------------------------------------------
# GET /mcp/audit/stats/by-tool
# ---------------------------------------------------------------------------

@router.get("/stats/by-tool", status_code=status.HTTP_200_OK)
async def audit_stats_by_tool(request: Request, days: int = 7):
    verify_internal_key(request)
    org_id = get_org_id(request)

    stats = await get_audit_stats_by_tool(org_id, days=days)
    return {"status": "success", "data": stats}


# ---------------------------------------------------------------------------
# GET /mcp/audit/stats/by-agent
# ---------------------------------------------------------------------------

@router.get("/stats/by-agent", status_code=status.HTTP_200_OK)
async def audit_stats_by_agent(request: Request, days: int = 7):
    verify_internal_key(request)
    org_id = get_org_id(request)

    stats = await get_audit_stats_by_agent(org_id, days=days)
    return {"status": "success", "data": stats}


# ---------------------------------------------------------------------------
# POST /mcp/audit/cleanup
# ---------------------------------------------------------------------------

@router.post("/cleanup", status_code=status.HTTP_200_OK)
async def cleanup_audit_logs(request: Request):
    """Delete audit logs older than retention period. Called by scheduled job."""
    verify_internal_key(request)
    from config import settings

    deleted = await delete_expired_audit_logs(settings.mcp_audit_retention_days)
    return {"status": "success", "deleted": deleted}


# ---------------------------------------------------------------------------
# POST /mcp/audit/cleanup-approvals
# ---------------------------------------------------------------------------

@router.post("/cleanup-approvals", status_code=status.HTTP_200_OK)
async def cleanup_approval_requests(request: Request):
    """Delete decided/expired approval requests older than retention period."""
    verify_internal_key(request)
    from config import settings
    from crud.mcp_approvals import delete_expired_approval_requests

    deleted = await delete_expired_approval_requests(settings.mcp_audit_retention_days)
    return {"status": "success", "deleted": deleted}