Spaces:
Running
Running
File size: 4,632 Bytes
3493993 | 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 | from __future__ import annotations
from typing import Any
from mcp.server.fastmcp import FastMCP
from app.analytics.schemas import AnalyticsQuery, AnalyticsSyncRequest
from app.mcp.registry import MCPRegistry
from app.security.context import auth_context
def register_analytics_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
"""Register narrow analytics tools over the shared domain service."""
def identity() -> tuple[str, str]:
context = auth_context.get()
return (
context.workspace_id or "" if context else "",
context.user_id or "" if context else "",
)
@server.tool(
name="analytics.overview", description="Get workspace or project analytics totals."
)
async def analytics_overview(filters: dict[str, Any] | None = None) -> dict[str, Any]:
async def action() -> dict[str, Any]:
workspace_id, _ = identity()
result = await registry.container.analytics.overview(
workspace_id, AnalyticsQuery.model_validate(filters or {})
)
return result.model_dump(mode="json")
return await registry.run_metadata_tool(
"analytics.overview", action, required_scope="analytics:read"
)
@server.tool(
name="analytics.timeseries", description="Get timezone-safe analytics time series."
)
async def analytics_timeseries(filters: dict[str, Any] | None = None) -> dict[str, Any]:
async def action() -> dict[str, Any]:
workspace_id, _ = identity()
result = await registry.container.analytics.timeseries(
workspace_id, AnalyticsQuery.model_validate(filters or {})
)
return result.model_dump(mode="json")
return await registry.run_metadata_tool(
"analytics.timeseries", action, required_scope="analytics:read"
)
@server.tool(name="analytics.platforms", description="Compare available provider metrics.")
async def analytics_platforms(filters: dict[str, Any] | None = None) -> dict[str, Any]:
return await analytics_overview(filters)
@server.tool(
name="analytics.top_content", description="List top content by an available metric."
)
async def analytics_top_content(filters: dict[str, Any] | None = None) -> dict[str, Any]:
async def action() -> dict[str, Any]:
workspace_id, _ = identity()
query = AnalyticsQuery.model_validate({**(filters or {}), "limit": 10})
result = await registry.container.analytics.posts(workspace_id, query)
return result.model_dump(mode="json")
return await registry.run_metadata_tool(
"analytics.top_content", action, required_scope="analytics:read"
)
@server.tool(name="analytics.post", description="Get analytics for one MediaRouter post.")
async def analytics_post(post_id: str, filters: dict[str, Any] | None = None) -> dict[str, Any]:
async def action() -> dict[str, Any]:
workspace_id, _ = identity()
result = await registry.container.analytics.post(
workspace_id, post_id, AnalyticsQuery.model_validate(filters or {})
)
return result.model_dump(mode="json")
return await registry.run_metadata_tool(
"analytics.post", action, required_scope="analytics:read"
)
@server.tool(
name="analytics.sync", description="Queue an idempotent analytics synchronization."
)
async def analytics_sync(payload: dict[str, Any]) -> dict[str, Any]:
async def action() -> dict[str, Any]:
workspace_id, user_id = identity()
result = await registry.container.analytics.create_sync(
workspace_id, user_id, AnalyticsSyncRequest.model_validate(payload)
)
return result.model_dump(mode="json")
return await registry.run_metadata_tool(
"analytics.sync", action, required_scope="analytics:sync"
)
@server.tool(name="analytics.sync_status", description="Get one analytics sync status.")
async def analytics_sync_status(sync_run_id: str) -> dict[str, Any]:
async def action() -> dict[str, Any]:
workspace_id, _ = identity()
run = await registry.container.analytics.repository.get_sync(workspace_id, sync_run_id)
return registry.container.analytics._sync_view(run).model_dump(mode="json")
return await registry.run_metadata_tool(
"analytics.sync_status", action, required_scope="analytics:read"
)
|