MCP Server for Coroot

Instructions

Implementation Reference

• src/mcp_coroot/server.py:629-656 (handler)

  MCP tool handler and registration for 'configure_integration'. This decorated function defines the tool interface, input schema via type hints, and delegates to the implementation.

  async def configure_integration(
      project_id: str,
      integration_type: str,
      config: dict[str, Any],
  ) -> dict[str, Any]:
      """Configure an integration for a project.
  
      Sets up or updates an integration configuration. Each integration
      type has specific configuration requirements.
  
      Integration types:
      - prometheus: Metrics data source
      - clickhouse: Long-term storage
      - aws: AWS services integration
      - slack: Slack notifications
      - teams: Microsoft Teams notifications
      - pagerduty: PagerDuty alerts
      - opsgenie: Opsgenie alerts
      - webhook: Custom webhooks
  
      Args:
          project_id: Project ID
          integration_type: Type of integration
          config: Integration-specific configuration dictionary
      """
      return await configure_integration_impl(  # type: ignore[no-any-return]
          project_id, integration_type, config
      )

• src/mcp_coroot/server.py:589-626 (helper)

  Internal helper that preprocesses webhook configurations and calls the client method, adding success response formatting.

  async def configure_integration_impl(
      project_id: str,
      integration_type: str,
      config: dict[str, Any],
  ) -> dict[str, Any]:
      """Configure an integration."""
      # Fix webhook configuration to include required templates
      if integration_type == "webhook":
          # Ensure required fields are present
          if "incidents" not in config:
              config["incidents"] = True
          if "deployments" not in config:
              config["deployments"] = False
  
          # Add default templates if missing
          if config.get("incidents") and "incident_template" not in config:
              config["incident_template"] = (
                  "Incident: {{.Title}}\n"
                  "Status: {{.Status}}\n"
                  "Applications: {{range .Applications}}{{.Id}} {{end}}\n"
                  "Link: {{.Link}}"
              )
          if config.get("deployments") and "deployment_template" not in config:
              config["deployment_template"] = (
                  "Deployment: {{.Application}}\n"
                  "Version: {{.Version}}\n"
                  "Status: {{.Status}}"
              )
  
      result = await get_client().configure_integration(
          project_id, integration_type, config
      )
      return {
          "success": True,
          "message": f"{integration_type} integration configured successfully",
          "integration": result,
      }

• src/mcp_coroot/client.py:613-649 (helper)

  Core CorootClient method that executes the HTTP PUT request to configure integrations via the Coroot API.

  async def configure_integration(
      self,
      project_id: str,
      integration_type: str,
      config: dict[str, Any],
  ) -> dict[str, Any]:
      """Configure an integration.
  
      Args:
          project_id: Project ID.
          integration_type: Type of integration (prometheus, slack, etc.).
          config: Integration-specific configuration.
  
      Returns:
          Updated integration configuration.
      """
      response = await self._request(
          "PUT",
          f"/api/project/{project_id}/integrations/{integration_type}",
          json=config,
      )
  
      # Handle different response types
      try:
          if response.headers.get("content-type", "").startswith("application/json"):
              data: dict[str, Any] = response.json()
              return data
          else:
              # If not JSON, return success with the provided config
              return {
                  "type": integration_type,
                  "config": config,
                  "status": "configured",
              }
      except Exception:
          # If parsing fails, return minimal success response
          return {"type": integration_type, "status": "configured"}