feat(seer-infra-telemetry): Add a PAT-only DatadogIdentityProvider

src/sentry/identity/__init__.py

@@ -11,7 +11,7 @@
 
 def _register_providers() -> None:
     from .bitbucket.provider import BitbucketIdentityProvider
-    from .datadog.provider import DatadogIdentityProvider
+    from .datadog.provider import DatadogIdentityProvider, DatadogPatIdentityProvider
     from .discord.provider import DiscordIdentityProvider
     from .gcp.provider import GCPIdentityProvider
     from .github.provider import GitHubIdentityProvider
@@ -36,6 +36,7 @@ def _register_providers() -> None:
     register(GoogleIdentityProvider)
     register(DiscordIdentityProvider)
     register(DatadogIdentityProvider)
+    register(DatadogPatIdentityProvider)
     register(GCPIdentityProvider)
 
 

src/sentry/identity/datadog/provider.py

@@ -13,6 +13,7 @@
 
 from sentry.auth.exceptions import IdentityNotValid
 from sentry.http import safe_urlopen, safe_urlread
+from sentry.identity.base import Provider
 from sentry.identity.mcp import McpIdentityProvider
 from sentry.identity.oauth2 import (
     OAuth2CallbackView,
@@ -401,3 +402,53 @@ def refresh_identity(self, identity: Identity | RpcIdentity, **kwargs: Any) -> N
         self.config["client_secret"] = client_secret
 
         super().refresh_identity(identity, **kwargs)
+
+
+class DatadogPatIdentityProvider(McpIdentityProvider, Provider):
+    """Datadog identity backed by a user-supplied read-only personal access token.
+
+    An alternative to the OAuth flow for environments where Datadog's MCP OAuth
+    (loopback-only redirect URIs) cannot be used. The submitted token is used as
+    a Bearer token against the MCP server, identical to an OAuth access token.
+    """
+
+    key = IntegrationProviderSlug.DATADOG_PAT
+    name = "Datadog (Personal Access Token)"
+
+    def get_pipeline_views(self) -> list[PipelineView[IdentityPipeline]]:
+        return []
+
+    def build_mcp_url(self, identity_data: dict[str, Any]) -> str | None:
+        """Full MCP endpoint URL for a stored Datadog identity.
+        Returns None when the site is missing or invalid."""
+        base = _mcp_base_url_for_site(identity_data.get("site"))
+        return f"{base}{MCP_ENDPOINT_PATH}" if base else None
+
+    def build_identity(self, data: dict[str, Any]) -> dict[str, Any]:
+        access_token = data.get("access_token")
+        if not access_token:
+            raise ValueError("Datadog requires an 'access_token' parameter.")
+
+        site = data.get("site")
+        base = _mcp_base_url_for_site(site)
+        if not site:
+            raise ValueError("Datadog requires a 'site' parameter (e.g. 'datadoghq.com').")
+        elif base is None:
+            raise ValueError(f"Invalid Datadog site: {site}")
+
+        user = get_user_info(access_token, base)
+        if "user_uuid" not in user or "org_uuid" not in user:
+            raise IdentityNotValid(
+                "User info response missing required fields (user_uuid, org_uuid)"
+            )
+
+        return {
+            "type": IntegrationProviderSlug.DATADOG_PAT,
+            "id": user["user_uuid"],
+            "idp_external_id": user["org_uuid"],
+            "idp_config": {"site": site},
+            "email": user.get("user_email"),
+            "name": user.get("user_name"),
+            "scopes": [],
+            "data": {"access_token": access_token, "site": site},
+        }

src/sentry/integrations/types.py

@@ -47,11 +47,16 @@ class IntegrationProviderSlug(StrEnum):
     OPSGENIE = "opsgenie"
     PERFORCE = "perforce"
     DATADOG = "datadog"
+    DATADOG_PAT = "datadog_pat"
     GCP = "gcp"
 
 
 MONITORING_PROVIDERS: frozenset[str] = frozenset(
-    {IntegrationProviderSlug.DATADOG, IntegrationProviderSlug.GCP}
+    {
+        IntegrationProviderSlug.DATADOG,
+        IntegrationProviderSlug.DATADOG_PAT,
+        IntegrationProviderSlug.GCP,
+    }
 )
 
 

src/sentry/seer/endpoints/seer_rpc.py

@@ -49,6 +49,7 @@
 from sentry.hybridcloud.rpc.service import RpcAuthenticationSetupException, RpcResolutionException
 from sentry.hybridcloud.rpc.sig import SerializableFunctionValueException
 from sentry.identity import default_manager as identity_manager
+from sentry.identity.oauth2 import OAuth2Provider
 from sentry.identity.services.identity import identity_service
 from sentry.integrations.github_enterprise.integration import GitHubEnterpriseIntegration
 from sentry.integrations.services.integration import integration_service
@@ -949,8 +950,12 @@ def refresh_monitoring_provider_token(
     if idp is None or idp.type not in MONITORING_PROVIDERS:
         return RefreshMonitoringProviderTokenErrorResponse(error="identity_not_found")
 
+    provider = identity_manager.get(idp.type)
+    if not isinstance(provider, OAuth2Provider):
+        # Static-token providers (e.g. Datadog PAT) have no refresh flow.
+        return RefreshMonitoringProviderTokenErrorResponse(error="refresh_not_supported")
+
     try:
-        provider = identity_manager.get(idp.type)
         provider.refresh_identity(identity)
     except IdentityNotValid:
         return RefreshMonitoringProviderTokenErrorResponse(error="identity_not_valid")

src/sentry/seer/sentry_data_models.py

@@ -885,7 +885,7 @@ def __contains__(self, key: object) -> bool:
 
 
 class RefreshMonitoringProviderTokenErrorResponse(BaseModel):
-    """`refresh_monitoring_provider_token` error: `{"error": <code>}`. The four
+    """`refresh_monitoring_provider_token` error: `{"error": <code>}`. The
     error codes the function emits — one per refusal branch — encoded as a
     Literal so the seer-side caller can switch on them safely."""
 
@@ -894,6 +894,7 @@ class RefreshMonitoringProviderTokenErrorResponse(BaseModel):
         "identity_not_found",
         "identity_not_valid",
         "refresh_failed",
+        "refresh_not_supported",
     ]
 
     def __getitem__(self, key: str) -> Any:

tests/sentry/identity/datadog/test_provider.py

@@ -23,6 +23,7 @@
     DatadogOAuth2CallbackView,
     DatadogOAuth2DCRView,
     DatadogOAuth2LoginView,
+    DatadogPatIdentityProvider,
     MissingPipelineStateError,
 )
 from sentry.identity.pipeline import IdentityPipeline
@@ -676,3 +677,78 @@ def test_build_mcp_url_missing_site(self) -> None:
 
     def test_build_mcp_url_invalid_site(self) -> None:
         assert self.provider.build_mcp_url({"site": "evil.example.com"}) is None
+
+
+@control_silo_test
+class DatadogPatIdentityProviderTest(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.provider = DatadogPatIdentityProvider()
+
+    def test_no_pipeline_views(self) -> None:
+        assert self.provider.get_pipeline_views() == []
+
+    @patch("sentry.identity.datadog.provider.get_user_info")
+    def test_build_identity(self, mock_get_user_info: MagicMock) -> None:
+        mock_get_user_info.return_value = {
+            "user_uuid": "dd-user-123",
+            "org_uuid": "dd-org-456",
+            "user_email": "user@example.com",
+            "user_name": "Test User",
+        }
+
+        result = self.provider.build_identity({"access_token": "pat-abc", "site": "datadoghq.com"})
+
+        assert result["type"] == "datadog_pat"
+        assert result["id"] == "dd-user-123"
+        assert result["idp_external_id"] == "dd-org-456"
+        assert result["idp_config"] == {"site": "datadoghq.com"}
+        assert result["email"] == "user@example.com"
+        assert result["name"] == "Test User"
+        assert result["scopes"] == []
+        assert result["data"] == {"access_token": "pat-abc", "site": "datadoghq.com"}
+        mock_get_user_info.assert_called_once_with("pat-abc", "https://mcp.datadoghq.com")
+
+    @patch("sentry.identity.datadog.provider.get_user_info")
+    def test_build_identity_missing_access_token(self, mock_get_user_info: MagicMock) -> None:
+        with pytest.raises(ValueError, match="requires an 'access_token'"):
+            self.provider.build_identity({"site": "datadoghq.com"})
+        mock_get_user_info.assert_not_called()
+
+    @patch("sentry.identity.datadog.provider.get_user_info")
+    def test_build_identity_missing_site(self, mock_get_user_info: MagicMock) -> None:
+        with pytest.raises(ValueError, match="requires a 'site'"):
+            self.provider.build_identity({"access_token": "pat-abc"})
+        mock_get_user_info.assert_not_called()
+
+    @patch("sentry.identity.datadog.provider.get_user_info")
+    def test_build_identity_invalid_site(self, mock_get_user_info: MagicMock) -> None:
+        with pytest.raises(ValueError, match="Invalid Datadog site"):
+            self.provider.build_identity({"access_token": "pat-abc", "site": "evil.example.com"})
+        mock_get_user_info.assert_not_called()
+
+    @patch("sentry.identity.datadog.provider.get_user_info")
+    def test_build_identity_missing_user_uuid(self, mock_get_user_info: MagicMock) -> None:
+        mock_get_user_info.return_value = {"org_uuid": "dd-org-456"}
+
+        with pytest.raises(IdentityNotValid, match="missing required fields"):
+            self.provider.build_identity({"access_token": "pat-abc", "site": "datadoghq.com"})
+
+    @patch("sentry.identity.datadog.provider.get_user_info")
+    def test_build_identity_missing_org_uuid(self, mock_get_user_info: MagicMock) -> None:
+        mock_get_user_info.return_value = {"user_uuid": "dd-user-123"}
+
+        with pytest.raises(IdentityNotValid, match="missing required fields"):
+            self.provider.build_identity({"access_token": "pat-abc", "site": "datadoghq.com"})
+
+    def test_build_mcp_url(self) -> None:
+        assert (
+            self.provider.build_mcp_url({"site": "datadoghq.com"})
+            == "https://mcp.datadoghq.com/api/unstable/mcp-server/mcp"
+        )
+
+    def test_build_mcp_url_missing_site(self) -> None:
+        assert self.provider.build_mcp_url({}) is None
+
+    def test_build_mcp_url_invalid_site(self) -> None:
+        assert self.provider.build_mcp_url({"site": "evil.example.com"}) is None

tests/sentry/seer/endpoints/test_seer_rpc.py

@@ -1825,6 +1825,20 @@ def test_missing_access_token_after_refresh(self) -> None:
         # Not "identity_not_valid" due to KeyError from get_oauth_data before reaching the .get() guard
         assert result == {"error": "refresh_failed"}
 
+    def test_pat_provider_not_refreshable(self) -> None:
+        # Static-token providers (Datadog PAT) have no refresh flow.
+        pat_idp = self.create_identity_provider(type="datadog_pat", external_id="dd-org-pat")
+        pat_identity = self.create_identity(
+            user=self.user,
+            identity_provider=pat_idp,
+            external_id="dd-user-pat",
+            data={"access_token": "pat-tok", "site": "datadoghq.com"},
+        )
+
+        result = refresh_monitoring_provider_token(identity_id=pat_identity.id)
+
+        assert result == {"error": "refresh_not_supported"}
+
 
 @with_feature("organizations:pr-metrics-attribution")
 @cell_silo_test