MCP Auditing

from tork_governance import TorkClient
from tork_governance.mcp import MCPAuditor

client = TorkClient()

# Create an MCP auditor
auditor = MCPAuditor(
    client=client,
    server_name="filesystem-server",
    log_level="detailed",  # "minimal", "standard", "detailed"
    policy_set="default"
)

# Wrap your MCP tool calls
@auditor.audit
async def read_file(path: str) -> str:
    # Your MCP tool implementation
    return await mcp_server.call_tool("read_file", {"path": path})

# mcp-policies.yaml
policies:
  - name: restrict-filesystem-access
    description: Limit file operations to specific directories
    tool_pattern: "filesystem:*"
    action: BLOCK
    conditions:
      - type: path_not_in
        allowed_paths:
          - "/app/data"
          - "/tmp"
    message: "File access outside allowed directories"

  - name: require-approval-for-writes
    description: Require human approval for write operations
    tool_pattern: "filesystem:write_file"
    action: APPROVAL
    approvers: ["admin@company.com"]

  - name: rate-limit-api-calls
    description: Limit external API calls
    tool_pattern: "http:*"
    action: BLOCK
    conditions:
      - type: rate_exceeded
        limit: 100
        window: "1m"

# Query audit logs
logs = client.mcp.get_audit_logs(
    server_name="filesystem-server",
    tool_name="read_file",
    start_time="2024-01-01T00:00:00Z",
    end_time="2024-01-31T23:59:59Z",
    status=["success", "blocked"]
)

for log in logs:
    print(f"{log.timestamp}: {log.tool_name}")
    print(f"  Agent: {log.agent_id}")
    print(f"  Status: {log.status}")
    print(f"  Duration: {log.duration_ms}ms")
    if log.policy_violations:
        print(f"  Violations: {log.policy_violations}")

# Get MCP metrics
metrics = client.mcp.get_metrics(
    server_name="filesystem-server",
    window="24h"
)

print(f"Total invocations: {metrics.total_invocations}")
print(f"Success rate: {metrics.success_rate}%")
print(f"Average latency: {metrics.avg_latency_ms}ms")
print(f"Policy blocks: {metrics.policy_blocks}")

# Tool breakdown
for tool in metrics.by_tool:
    print(f"  {tool.name}: {tool.invocations} calls")