test_integration
Verify integration configuration and connectivity for Coroot observability platform projects. Tests connections to services like Prometheus or Slack.
Instructions
Test an integration configuration.
Verifies that an integration is properly configured and can connect.
Args: project_id: Project ID integration_type: Type of integration (prometheus, slack, etc)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| project_id | Yes | ||
| integration_type | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/mcp_coroot/server.py:1421-1432 (registration)Registration of the 'test_integration' MCP tool via @mcp.tool() decorator, with handler function that wraps the implementation.
@mcp.tool() async def test_integration(project_id: str, integration_type: str) -> dict[str, Any]: """Test an integration configuration. Verifies that an integration is properly configured and can connect. Args: project_id: Project ID integration_type: Type of integration (prometheus, slack, etc) """ return await test_integration_impl(project_id, integration_type) # type: ignore[no-any-return] - src/mcp_coroot/server.py:1408-1419 (handler)Wrapper handler implementation that calls the CorootClient.test_integration method and formats the response.
@handle_errors async def test_integration_impl( project_id: str, integration_type: str ) -> dict[str, Any]: """Test an integration.""" result = await get_client().test_integration(project_id, integration_type) return { "success": True, "message": f"{integration_type} integration test completed", "result": result, } - src/mcp_coroot/client.py:1036-1059 (helper)Core implementation in CorootClient that fetches current integration config and sends a test POST request to the Coroot API.
async def test_integration( self, project_id: str, integration_type: str ) -> dict[str, Any]: """Test an integration configuration. Args: project_id: Project ID. integration_type: Type of integration to test. Returns: Test results. """ # First get the current config integrations = await self.list_integrations(project_id) current_config = integrations.get(integration_type, {}) # Send POST with current config to test it response = await self._request( "POST", f"/api/project/{project_id}/integrations/{integration_type}", json=current_config, ) data: dict[str, Any] = response.json() return data