Tomas Kracmar
5122739c01
All checks were successful
CI / lint-and-test (push) Successful in 36s
- Extract shared MCP tool handlers to mcp_common.py - mcp_server.py now uses shared handlers (stdio transport for local dev) - New routes/mcp.py: SSE transport behind existing OIDC Bearer auth - Mount MCP ASGI app at /mcp in main.py when AI_FEATURES_ENABLED - /mcp/sse -> establishes SSE stream (requires valid token when auth enabled) - /mcp/messages/ -> receives MCP client messages - Update README with SSE MCP docs - Add tests for mount existence, auth, and message routing
89 lines
2.4 KiB
Python
89 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
AOC MCP Server — stdio transport
|
|
|
|
Standalone MCP server for local use (Claude Desktop, Cursor, etc.).
|
|
For the HTTP/SSE version (production, behind auth), see routes/mcp.py.
|
|
|
|
Usage:
|
|
python mcp_server.py
|
|
|
|
Claude Desktop config (~/.config/claude/claude_desktop_config.json):
|
|
{
|
|
"mcpServers": {
|
|
"aoc": {
|
|
"command": "python",
|
|
"args": ["/path/to/aoc/backend/mcp_server.py"],
|
|
"env": {"MONGO_URI": "mongodb://..."}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
|
|
# Ensure backend modules are importable when run standalone
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from mcp.server import Server
|
|
from mcp.server.stdio import stdio_server
|
|
from mcp.types import TextContent, Tool
|
|
from mcp_common import (
|
|
ASK_SCHEMA,
|
|
GET_EVENT_SCHEMA,
|
|
GET_SUMMARY_SCHEMA,
|
|
SEARCH_EVENTS_SCHEMA,
|
|
handle_ask,
|
|
handle_get_event,
|
|
handle_get_summary,
|
|
handle_search_events,
|
|
)
|
|
|
|
app = Server("aoc")
|
|
|
|
|
|
@app.list_tools()
|
|
async def list_tools() -> list[Tool]:
|
|
return [
|
|
Tool(
|
|
name="search_events",
|
|
description="Search audit events by entity, service, operation, or result.",
|
|
inputSchema=SEARCH_EVENTS_SCHEMA,
|
|
),
|
|
Tool(name="get_event", description="Retrieve a single audit event by its ID.", inputSchema=GET_EVENT_SCHEMA),
|
|
Tool(
|
|
name="get_summary",
|
|
description="Get an aggregated summary of audit activity for the last N days.",
|
|
inputSchema=GET_SUMMARY_SCHEMA,
|
|
),
|
|
Tool(
|
|
name="ask",
|
|
description="Ask a natural language question about audit logs. Returns a narrative answer.",
|
|
inputSchema=ASK_SCHEMA,
|
|
),
|
|
]
|
|
|
|
|
|
@app.call_tool()
|
|
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
|
|
if name == "search_events":
|
|
return await handle_search_events(arguments)
|
|
if name == "get_event":
|
|
return await handle_get_event(arguments)
|
|
if name == "get_summary":
|
|
return await handle_get_summary(arguments)
|
|
if name == "ask":
|
|
return await handle_ask(arguments)
|
|
raise ValueError(f"Unknown tool: {name}")
|
|
|
|
|
|
async def main():
|
|
async with stdio_server() as (read_stream, write_stream):
|
|
await app.run(read_stream, write_stream, app.create_initialization_options())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|