MCP Server for Coroot

Instructions

Implementation Reference

• src/mcp_coroot/server.py:356-384 (handler)

  Primary MCP tool handler for 'get_application_traces', decorated with @mcp.tool() for automatic registration and schema inference from signature. Delegates to internal impl.
  @mcp.tool() async def get_application_traces( project_id: str, app_id: str, from_timestamp: int | None = None, to_timestamp: int | None = None, trace_id: str | None = None, query: str | None = None, ) -> dict[str, Any]: """Get distributed traces for an application. Retrieves distributed tracing data showing request flow through the application and its dependencies. ⚠️ WARNING: This endpoint can return very large responses (100k+ tokens) when retrieving many traces. Consider using time filters or trace_id to limit the response size. Args: project_id: Project ID app_id: Application ID (format: namespace/kind/name) from_timestamp: Start timestamp (optional, recommended to limit data) to_timestamp: End timestamp (optional, recommended to limit data) trace_id: Specific trace ID to retrieve (optional, returns single trace) query: Search query (optional) """ return await get_application_traces_impl( # type: ignore[no-any-return] project_id, app_id, from_timestamp, to_timestamp, trace_id, query )
• src/mcp_coroot/server.py:334-354 (helper)

  Internal helper implementation that encodes app_id and calls CorootClient.get_application_traces
  @handle_errors async def get_application_traces_impl( project_id: str, app_id: str, from_timestamp: int | None = None, to_timestamp: int | None = None, trace_id: str | None = None, query: str | None = None, ) -> dict[str, Any]: """Get application traces.""" # URL encode the app_id since it contains slashes encoded_app_id = quote(app_id, safe="") traces = await get_client().get_application_traces( project_id, encoded_app_id, from_timestamp, to_timestamp, trace_id, query ) return { "success": True, "traces": traces, }
• src/mcp_coroot/client.py:426-464 (helper)

  CorootClient method implementing the core logic: constructs query params and makes HTTP GET request to Coroot API endpoint /api/project/{project_id}/app/{app_id}/tracing to fetch distributed traces data.
  async def get_application_traces( self, project_id: str, app_id: str, from_timestamp: int | None = None, to_timestamp: int | None = None, trace_id: str | None = None, query: str | None = None, ) -> dict[str, Any]: """Get application distributed traces. Args: project_id: Project ID. app_id: Application ID. from_timestamp: Start timestamp. to_timestamp: End timestamp. trace_id: Specific trace ID. query: Search query. Returns: Distributed traces data. """ params = {} if from_timestamp: params["from"] = str(from_timestamp) if to_timestamp: params["to"] = str(to_timestamp) if trace_id: params["trace_id"] = trace_id if query: params["query"] = query response = await self._request( "GET", f"/api/project/{project_id}/app/{app_id}/tracing", params=params, ) data: dict[str, Any] = response.json() return data
• src/mcp_coroot/server.py:356-356 (registration)

  @mcp.tool() decorator registers the function as an MCP tool, auto-generates schema from signature.
  @mcp.tool()