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"}