MCP Server for Coroot

Instructions

Implementation Reference

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

  Core handler function in CorootClient class that executes the HTTP PUT request to the Coroot API endpoint for configuring integrations. Handles JSON parsing and provides fallback responses.
  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"}
• src/mcp_coroot/server.py:628-657 (registration)

  MCP tool registration for 'configure_integration' using FastMCP @mcp.tool() decorator. Defines input parameters, docstring with supported integration types, and calls the implementation function.
  @mcp.tool() 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)

  Server-side helper function that prepares the configuration (adds defaults for webhook templates) and calls the CorootClient.configure_integration method. Wraps @handle_errors decorator.
  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, }