feat(claude-code): Add API-driven pipeline backend for Claude Code integration setup

src/sentry/integrations/claude_code/integration.py

@@ -14,9 +14,12 @@
 
 from django import forms
 from django.conf import settings as django_settings
+from django.http.request import HttpRequest
 from django.utils.translation import gettext_lazy as _
 from pydantic import BaseModel, validator
+from rest_framework.fields import CharField
 
+from sentry.api.serializers.rest_framework.base import CamelSnakeSerializer
 from sentry.integrations.base import (
     FeatureDescription,
     IntegrationData,
@@ -32,6 +35,8 @@
 from sentry.integrations.models.integration import Integration
 from sentry.integrations.pipeline import IntegrationPipeline
 from sentry.integrations.services.integration.model import RpcIntegration
+from sentry.pipeline.types import PipelineStepResult
+from sentry.pipeline.views.base import ApiPipelineSteps
 from sentry.seer.autofix.utils import CodingAgentState
 from sentry.shared_integrations.exceptions import IntegrationConfigurationError
 from sentry.utils.imports import import_string
@@ -153,6 +158,29 @@ def bind_state(self, pipeline: IntegrationPipeline, form: forms.Form) -> None:
         pipeline.bind_state(self.get_state_key(), form.cleaned_data["api_key"])
 
 
+class ClaudeCodeApiKeySerializer(CamelSnakeSerializer):
+    api_key = CharField(required=True, max_length=255)
+
+
+class ClaudeCodeApiKeyApiStep:
+    step_name = "api_key_config"
+
+    def get_step_data(self, pipeline: IntegrationPipeline, request: HttpRequest) -> dict[str, Any]:
+        return {}
+
+    def get_serializer_cls(self) -> type:
+        return ClaudeCodeApiKeySerializer
+
+    def handle_post(
+        self,
+        validated_data: dict[str, str],
+        pipeline: IntegrationPipeline,
+        request: HttpRequest,
+    ) -> PipelineStepResult:
+        pipeline.bind_state("api_key", validated_data["api_key"])
+        return PipelineStepResult.advance()
+
+
 class ClaudeCodeAgentIntegrationProvider(CodingAgentIntegrationProvider):
     """
     Integration provider for Claude Code Agent.
@@ -168,6 +196,9 @@ class ClaudeCodeAgentIntegrationProvider(CodingAgentIntegrationProvider):
     def get_pipeline_views(self):
         return [ClaudeCodeApiKeyPipelineView()]
 
+    def get_pipeline_api_steps(self) -> ApiPipelineSteps[IntegrationPipeline]:
+        return [ClaudeCodeApiKeyApiStep()]
+
     def build_integration(self, state: Mapping[str, Any]) -> IntegrationData:
         api_key = state.get("api_key")
         if not api_key:

tests/sentry/integrations/claude_code/test_integration.py

@@ -9,15 +9,21 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from django.urls import reverse
 
 from sentry.integrations.claude_code.integration import (
     PROVIDER_KEY,
     PROVIDER_NAME,
     ClaudeCodeAgentIntegration,
     ClaudeCodeAgentIntegrationProvider,
 )
+from sentry.integrations.models.integration import Integration
+from sentry.integrations.models.organization_integration import OrganizationIntegration
+from sentry.integrations.pipeline import IntegrationPipeline
 from sentry.shared_integrations.exceptions import IntegrationConfigurationError
-from sentry.testutils.cases import IntegrationTestCase
+from sentry.testutils.cases import APITestCase, IntegrationTestCase
+from sentry.testutils.helpers.features import with_feature
+from sentry.testutils.silo import control_silo_test
 
 MOCK_GET_CLIENT_CLASS = "sentry.integrations.claude_code.integration._get_client_class"
 
@@ -493,3 +499,66 @@ def test_launch_updates_both_environment_and_agent_when_changed(self) -> None:
         assert installation.model.metadata["environment_id"] == "env-new"
         assert installation.model.metadata["agent_id"] == "agent-new"
         assert installation.model.metadata["agent_version"] == 2
+
+
+@control_silo_test
+class ClaudeCodeApiPipelineTest(APITestCase):
+    endpoint = "sentry-api-0-organization-pipeline"
+    method = "post"
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.login_as(self.user)
+
+    def _get_pipeline_url(self) -> str:
+        return reverse(
+            self.endpoint,
+            args=[self.organization.slug, IntegrationPipeline.pipeline_name],
+        )
+
+    def _initialize_pipeline(self) -> Any:
+        return self.client.post(
+            self._get_pipeline_url(),
+            data={"action": "initialize", "provider": "claude_code"},
+            format="json",
+        )
+
+    def _advance_step(self, data: dict[str, Any]) -> Any:
+        return self.client.post(self._get_pipeline_url(), data=data, format="json")
+
+    @with_feature("organizations:integrations-claude-code")
+    def test_initialize_pipeline(self) -> None:
+        resp = self._initialize_pipeline()
+        assert resp.status_code == 200
+        assert resp.data["step"] == "api_key_config"
+        assert resp.data["stepIndex"] == 0
+        assert resp.data["totalSteps"] == 1
+        assert resp.data["provider"] == "claude_code"
+
+    @with_feature("organizations:integrations-claude-code")
+    def test_missing_api_key(self) -> None:
+        self._initialize_pipeline()
+        resp = self._advance_step({})
+        assert resp.status_code == 400
+
+    @with_feature("organizations:integrations-claude-code")
+    def test_full_pipeline_flow(self) -> None:
+        mock_cls, _ = _mock_client_class()
+
+        resp = self._initialize_pipeline()
+        assert resp.data["step"] == "api_key_config"
+
+        with patch(MOCK_GET_CLIENT_CLASS, return_value=mock_cls):
+            resp = self._advance_step({"apiKey": "sk-ant-test-key-123"})
+
+        assert resp.status_code == 200
+        assert resp.data["status"] == "complete"
+
+        integration = Integration.objects.get(provider="claude_code")
+        assert integration.name == PROVIDER_NAME
+        assert integration.metadata["api_key"] == "sk-ant-test-key-123"
+
+        assert OrganizationIntegration.objects.filter(
+            organization_id=self.organization.id,
+            integration=integration,
+        ).exists()