MCP Server for Coroot

Instructions

Implementation Reference

• src/mcp_coroot/server.py:334-354 (handler)

  Core handler implementation that encodes app_id, calls CorootClient.get_application_traces(), and formats the response.
  @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/server.py:356-385 (registration)

  MCP tool registration using @mcp.tool() decorator. Defines input parameters and comprehensive docstring serving as schema. Delegates to 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/client.py:426-464 (handler)

  CorootClient method implementing the HTTP request to fetch distributed traces from Coroot API endpoint.
  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:44-87 (helper)

  Error handling decorator applied to all MCP tools, including get_application_traces_impl, standardizes error responses.
  def handle_errors(func: Callable[..., Any]) -> Callable[..., Any]: """Decorator to handle common errors in tool implementations.""" @wraps(func) async def wrapper(*args: Any, **kwargs: Any) -> dict[str, Any]: try: return await func(*args, **kwargs) # type: ignore[no-any-return] except ValueError as e: # Handle missing credentials or other validation errors return { "success": False, "error": str(e), "error_type": "validation", } except CorootError as e: # Handle Coroot API errors (including authentication) error_msg = str(e) if "Authentication failed" in error_msg: return { "success": False, "error": error_msg, "error_type": "authentication", } elif "401" in error_msg or "Unauthorized" in error_msg: return { "success": False, "error": "Authentication required. Please check your credentials.", "error_type": "authentication", } return { "success": False, "error": error_msg, "error_type": "api_error", } except Exception as e: # Handle unexpected errors return { "success": False, "error": f"Unexpected error: {str(e)}", "error_type": "unknown", } return wrapper
• src/mcp_coroot/server.py:93-113 (helper)

  Lazy singleton initialization of CorootClient used by all tools to make authenticated API calls.
  def get_client() -> CorootClient: """Get or create the client instance. Raises: ValueError: If no credentials are configured. """ global _client if _client is None: try: _client = CorootClient() except ValueError as e: # Re-raise with more context raise ValueError( "Coroot credentials not configured. " "Please set COROOT_BASE_URL and either:\n" " - COROOT_USERNAME and COROOT_PASSWORD for automatic login\n" " - COROOT_SESSION_COOKIE for direct authentication\n" " - COROOT_API_KEY for data ingestion endpoints" ) from e return _client