MCP Server for Coroot

Instructions

Implementation Reference

• src/mcp_coroot/server.py:1166-1179 (handler)

  MCP tool handler for 'get_db_instrumentation'. This is the entrypoint function decorated with @mcp.tool() that executes the tool logic by delegating to the implementation helper.

  @mcp.tool()
  async def get_db_instrumentation(
      project_id: str, app_id: str, db_type: str
  ) -> dict[str, Any]:
      """Get database instrumentation configuration.
  
      Retrieves instrumentation settings for a specific database type.
  
      Args:
          project_id: Project ID
          app_id: Application ID (format: namespace/kind/name)
          db_type: Database type (mysql, postgres, redis, mongodb, memcached)
      """
      return await get_db_instrumentation_impl(project_id, app_id, db_type)  # type: ignore[no-any-return]

• src/mcp_coroot/server.py:1154-1164 (helper)

  Helper implementation that wraps the CorootClient call and formats the response with success indicator.

  async def get_db_instrumentation_impl(
      project_id: str, app_id: str, db_type: str
  ) -> dict[str, Any]:
      """Get database instrumentation config."""
      client = get_client()
      config = await client.get_db_instrumentation(project_id, app_id, db_type)
      return {
          "success": True,
          "config": config,
      }

• src/mcp_coroot/client.py:1384-1408 (helper)

  Underlying CorootClient method that makes the actual HTTP API request to retrieve DB instrumentation configuration.

  async def get_db_instrumentation(
      self, project_id: str, app_id: str, db_type: str
  ) -> dict[str, Any]:
      """Get database instrumentation configuration.
  
      Args:
          project_id: Project ID.
          app_id: Application ID.
          db_type: Database type (mysql, postgres, redis, mongodb, memcached).
  
      Returns:
          Database instrumentation configuration.
      """
      # URL encode the app_id since it contains slashes
      from urllib.parse import quote
  
      encoded_app_id = quote(app_id, safe="")
  
      response = await self._request(
          "GET",
          f"/api/project/{project_id}/app/{encoded_app_id}/instrumentation/{db_type}",
      )
      data: dict[str, Any] = response.json()
      return data