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()