MCP Server for Coroot

Instructions

Implementation Reference

• src/mcp_coroot/server.py:867-914 (handler)

  Core handler logic for the MCP tool 'update_application_category'. Fetches current category config, merges updates, and calls Coroot API to persist changes.

  async def update_application_category_impl(
      project_id: str,
      name: str,
      custom_patterns: str | None = None,
      notify_incidents: bool | None = None,
      notify_deployments: bool | None = None,
      slack_channel: str | None = None,
  ) -> dict[str, Any]:
      """Update an existing application category."""
      # First get the existing category
      categories = await get_client().get_application_categories(project_id)
      existing: dict[str, Any] | None = None
      for cat in categories:
          if isinstance(cat, dict) and cat.get("name") == name:
              existing = cat
              break
  
      if not existing:
          raise ValueError(f"Category '{name}' not found")
  
      # Update only specified fields
      if custom_patterns is not None:
          existing["custom_patterns"] = custom_patterns
  
      if notify_incidents is not None:
          existing["notification_settings"]["incidents"]["enabled"] = notify_incidents
  
      if notify_deployments is not None:
          existing["notification_settings"]["deployments"]["enabled"] = notify_deployments
  
      if slack_channel is not None:
          if "slack" not in existing["notification_settings"]["incidents"]:
              existing["notification_settings"]["incidents"]["slack"] = {}
          slack = existing["notification_settings"]["incidents"]["slack"]
          slack["channel"] = slack_channel
          slack["enabled"] = True
  
          if "slack" not in existing["notification_settings"]["deployments"]:
              existing["notification_settings"]["deployments"]["slack"] = {}
          slack = existing["notification_settings"]["deployments"]["slack"]
          slack["channel"] = slack_channel
  
      await get_client().update_application_category(project_id, name, existing)
      return {
          "success": True,
          "message": f"Application category '{name}' updated successfully",
      }

• src/mcp_coroot/server.py:916-946 (registration)

  MCP tool registration via @mcp.tool() decorator. Defines input parameters and docstring schema for the 'update_application_category' tool.

  @mcp.tool()
  async def update_application_category(
      project_id: str,
      name: str,
      custom_patterns: str | None = None,
      notify_incidents: bool | None = None,
      notify_deployments: bool | None = None,
      slack_channel: str | None = None,
  ) -> dict[str, Any]:
      """Update an existing application category.
  
      Updates specific fields of an application category.
      Only provided fields are updated.
  
      Args:
          project_id: Project ID
          name: Category name to update
          custom_patterns: New space-separated glob patterns (optional)
          notify_incidents: Whether to notify about incidents (optional)
          notify_deployments: Whether to notify about deployments (optional)
          slack_channel: Slack channel for notifications (optional)
      """
      return await update_application_category_impl(  # type: ignore[no-any-return]
          project_id,
          name,
          custom_patterns,
          notify_incidents,
          notify_deployments,
          slack_channel,
      )

• src/mcp_coroot/client.py:749-768 (helper)

  CorootClient helper method that performs the HTTP POST request to the Coroot API endpoint for updating application categories.

  async def update_application_category(
      self, project_id: str, category_name: str, category: dict[str, Any]
  ) -> dict[str, Any]:
      """Update an existing application category.
  
      Args:
          project_id: Project ID.
          category_name: Name of the category to update.
          category: Updated category object.
  
      Returns:
          Updated category.
      """
      # Set the id field for updates
      category["id"] = category_name
      response = await self._request(
          "POST", f"/api/project/{project_id}/application_categories", json=category
      )
      return self._parse_json_response(response)

• src/mcp_coroot/server.py:93-113 (helper)

  Shared helper function that provides the authenticated CorootClient instance used by all MCP tools.

  def get_client() -> CorootClient:
      """Get or create the client instance.
  
      Raises:
          ValueError: If no credentials are configured.
      """
      global _client
      if _client is None:
          try:
              _client = CorootClient()
          except ValueError as e:
              # Re-raise with more context
              raise ValueError(
                  "Coroot credentials not configured. "
                  "Please set COROOT_BASE_URL and either:\n"
                  "  - COROOT_USERNAME and COROOT_PASSWORD for automatic login\n"
                  "  - COROOT_SESSION_COOKIE for direct authentication\n"
                  "  - COROOT_API_KEY for data ingestion endpoints"
              ) from e
      return _client

• src/mcp_coroot/server.py:44-87 (helper)

  Error handling decorator applied to the handler implementation for standardized error responses.

  def handle_errors(func: Callable[..., Any]) -> Callable[..., Any]:
      """Decorator to handle common errors in tool implementations."""
  
      @wraps(func)
      async def wrapper(*args: Any, **kwargs: Any) -> dict[str, Any]:
          try:
              return await func(*args, **kwargs)  # type: ignore[no-any-return]
          except ValueError as e:
              # Handle missing credentials or other validation errors
              return {
                  "success": False,
                  "error": str(e),
                  "error_type": "validation",
              }
          except CorootError as e:
              # Handle Coroot API errors (including authentication)
              error_msg = str(e)
              if "Authentication failed" in error_msg:
                  return {
                      "success": False,
                      "error": error_msg,
                      "error_type": "authentication",
                  }
              elif "401" in error_msg or "Unauthorized" in error_msg:
                  return {
                      "success": False,
                      "error": "Authentication required. Please check your credentials.",
                      "error_type": "authentication",
                  }
              return {
                  "success": False,
                  "error": error_msg,
                  "error_type": "api_error",
              }
          except Exception as e:
              # Handle unexpected errors
              return {
                  "success": False,
                  "error": f"Unexpected error: {str(e)}",
                  "error_type": "unknown",
              }
  
      return wrapper