feat(management): Flux API key bearer tokens

src/foxnose_sdk/__init__.py

@@ -73,6 +73,7 @@
     FolderList,
     FolderSummary,
     FluxAPIKeyList,
+    FluxAPIKeyBearerToken,
     FluxAPIKeySummary,
     ManagementAPIKeyList,
     ManagementAPIKeySummary,
@@ -163,6 +164,7 @@
     "EnvironmentList",
     "ManagementAPIKeySummary",
     "ManagementAPIKeyList",
+    "FluxAPIKeyBearerToken",
     "FluxAPIKeySummary",
     "FluxAPIKeyList",
     "ManagementRoleSummary",

src/foxnose_sdk/management/client.py

@@ -28,6 +28,7 @@
     FieldList,
     FieldSummary,
     FluxAPIKeyList,
+    FluxAPIKeyBearerToken,
     FluxAPIKeySummary,
     FluxRoleList,
     FluxRoleSummary,
@@ -238,6 +239,9 @@ def _flux_api_keys_root(self) -> str:
     def _flux_api_key_root(self, api_key: str) -> str:
         return f"{self._flux_api_keys_root()}/{api_key}"
 
+    def _flux_api_key_bearer_token_root(self, api_key: str) -> str:
+        return f"{self._flux_api_key_root(api_key)}/bearer-token"
+
     # API management paths
     def _apis_root(self) -> str:
         return f"/v1/{self.environment_key}/api"
@@ -547,6 +551,53 @@ def delete_flux_api_key(self, key: FluxAPIKeyRef) -> None:
         key = _resolve_key(key)
         self.request("DELETE", f"{self._flux_api_key_root(key)}/", parse_json=False)
 
+    def issue_flux_api_key_bearer_token(
+        self, key: FluxAPIKeyRef
+    ) -> FluxAPIKeyBearerToken:
+        """Issue a bearer token for a Flux API key, or replace the existing one.
+
+        A bearer token exists for clients that accept a single token value and
+        send it as ``Authorization: Bearer <token>`` with no way to choose the
+        scheme — hosted MCP connectors, notably the Claude API's
+        ``mcp_servers``. Those clients cannot send ``Simple`` or ``Secure`` at
+        all, which put every authentication-required Flux API out of their reach.
+
+        THE PLAINTEXT IS RETURNED ONLY HERE, AND ONLY ONCE. The service stores a
+        hash, exactly as it does for ``secret_key``, so a lost token is re-issued
+        rather than recovered. Later key reads expose only
+        ``bearer_token_prefix``.
+
+        The key itself is untouched: ``public_key``, ``secret_key``, ``role`` and
+        grants all survive, so ``Simple`` and ``Secure`` integrations keep
+        working across a re-issue. That is what makes this the way to cut off a
+        connector without recreating a key and reconfiguring everything using it.
+
+        There is at most one token per key; calling this again replaces it.
+
+        Args:
+            key: Unique identifier of the API key.
+        """
+        key = _resolve_key(key)
+        data = self.request("POST", f"{self._flux_api_key_bearer_token_root(key)}/")
+        return FluxAPIKeyBearerToken.model_validate(data)
+
+    def revoke_flux_api_key_bearer_token(self, key: FluxAPIKeyRef) -> None:
+        """Revoke a Flux API key's bearer token. The key keeps working.
+
+        Only the token is removed — the key, its role and its ``Simple`` /
+        ``Secure`` credentials are untouched. Idempotent: succeeds whether or not
+        a token was issued.
+
+        Args:
+            key: Unique identifier of the API key.
+        """
+        key = _resolve_key(key)
+        self.request(
+            "DELETE",
+            f"{self._flux_api_key_bearer_token_root(key)}/",
+            parse_json=False,
+        )
+
     # ------------------------------------------------------------------ #
     # API management operations
     # ------------------------------------------------------------------ #
@@ -2966,6 +3017,23 @@ async def delete_flux_api_key(self, key: FluxAPIKeyRef) -> None:
             "DELETE", f"{self._flux_api_key_root(key)}/", parse_json=False
         )
 
+    async def issue_flux_api_key_bearer_token(
+        self, key: FluxAPIKeyRef
+    ) -> FluxAPIKeyBearerToken:
+        key = _resolve_key(key)
+        data = await self.request(
+            "POST", f"{self._flux_api_key_bearer_token_root(key)}/"
+        )
+        return FluxAPIKeyBearerToken.model_validate(data)
+
+    async def revoke_flux_api_key_bearer_token(self, key: FluxAPIKeyRef) -> None:
+        key = _resolve_key(key)
+        await self.request(
+            "DELETE",
+            f"{self._flux_api_key_bearer_token_root(key)}/",
+            parse_json=False,
+        )
+
     # ------------------------------------------------------------------ #
     # API management operations (async)
     # ------------------------------------------------------------------ #

src/foxnose_sdk/management/models.py

@@ -311,11 +311,37 @@ class FluxAPIKeySummary(BaseModel):
     role: str | None = None
     environment: str
     created_at: datetime
+    #: First 12 characters of the key's bearer token (e.g. ``fxk_A7fQ2mXe``), or
+    #: None when none is issued. Enough to recognise a token in a config file or
+    #: a log; never enough to use one — the token itself is returned only by
+    #: :meth:`ManagementClient.issue_flux_api_key_bearer_token`, once.
+    #: Optional so the model still validates against a server predating the
+    #: feature.
+    bearer_token_prefix: str | None = None
+    #: When the current bearer token was issued, or None.
+    bearer_token_issued_at: datetime | None = None
 
 
 FluxAPIKeyList = PaginatedResponse[FluxAPIKeySummary]
 
 
+class FluxAPIKeyBearerToken(BaseModel):
+    """The one-time response from issuing or re-issuing a bearer token.
+
+    A bearer token is an opaque credential bound to a Flux API key, for hosted
+    MCP connectors that accept only a token value and send it as
+    ``Authorization: Bearer <token>``. It identifies the key and nothing more:
+    role, grants and per-collection permissions are the key's own.
+    """
+
+    #: The credential, e.g. ``fxk_A7fQ2mXe...``. RETURNED ONLY HERE, ONLY ONCE —
+    #: the service stores a hash, so a lost token is re-issued, not recovered.
+    bearer_token: str
+    #: First 12 characters, also present on every subsequent key read.
+    bearer_token_prefix: str
+    bearer_token_issued_at: datetime
+
+
 class ManagementRoleSummary(BaseModel):
     """Represents a management API role."""
 

tests/test_async_clients.py

@@ -579,6 +579,41 @@ def handler(request: httpx.Request) -> httpx.Response:
     await client.aclose()
 
 
+@pytest.mark.asyncio
+async def test_async_flux_api_key_bearer_token_lifecycle():
+    """Same contract as the sync client: the sub-resource, never the key."""
+    captured: dict[str, Any] = {"paths": []}
+    token_json = {
+        "bearer_token": "fxk_A7fQ2mXeKp3vR8sT1uW5yZ2bC6dF9gH0jL4nQ7x",
+        "bearer_token_prefix": "fxk_A7fQ2mXe",
+        "bearer_token_issued_at": "2026-08-09T10:24:11.482Z",
+    }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["paths"].append((request.method, request.url.path))
+        if request.method == "POST":
+            return httpx.Response(200, json=token_json)
+        if request.method == "DELETE":
+            return httpx.Response(204)
+        raise AssertionError("Unexpected call")
+
+    client = build_async_management_client(handler)
+
+    issued = await client.issue_flux_api_key_bearer_token("flux-key-1")
+    assert issued.bearer_token == token_json["bearer_token"]
+    assert captured["paths"][0] == (
+        "POST",
+        "/v1/env123/permissions/flux-api/api-keys/flux-key-1/bearer-token/",
+    )
+
+    await client.revoke_flux_api_key_bearer_token("flux-key-1")
+    method, path = captured["paths"][-1]
+    assert method == "DELETE"
+    assert path.endswith("/api-keys/flux-key-1/bearer-token/")
+    assert not path.endswith("/api-keys/flux-key-1/")
+    await client.aclose()
+
+
 @pytest.mark.asyncio
 async def test_async_management_role_crud():
     captured: list[str] = []

tests/test_clients.py

@@ -673,6 +673,65 @@ def handler(request: httpx.Request) -> httpx.Response:
     assert captured["paths"][-1][0] == "DELETE"
 
 
+def test_flux_api_key_bearer_token_lifecycle():
+    """Issue and revoke address the SUB-RESOURCE, never the key itself.
+
+    A request to the key's own URL would delete the key and take its
+    Simple/Secure credentials with it — the token is a separate credential
+    precisely so it can be replaced without disturbing them.
+    """
+    captured: dict[str, Any] = {"paths": []}
+    token_json = {
+        "bearer_token": "fxk_A7fQ2mXeKp3vR8sT1uW5yZ2bC6dF9gH0jL4nQ7x",
+        "bearer_token_prefix": "fxk_A7fQ2mXe",
+        "bearer_token_issued_at": "2026-08-09T10:24:11.482Z",
+    }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["paths"].append((request.method, request.url.path))
+        if request.method == "POST":
+            return httpx.Response(200, json=token_json)
+        if request.method == "DELETE":
+            return httpx.Response(204)
+        raise AssertionError("Unexpected call")
+
+    client = build_management_client(handler)
+
+    issued = client.issue_flux_api_key_bearer_token("flux-key-1")
+    assert issued.bearer_token == token_json["bearer_token"]
+    assert issued.bearer_token_prefix == "fxk_A7fQ2mXe"
+    assert captured["paths"][0] == (
+        "POST",
+        "/v1/env123/permissions/flux-api/api-keys/flux-key-1/bearer-token/",
+    )
+
+    client.revoke_flux_api_key_bearer_token("flux-key-1")
+    method, path = captured["paths"][-1]
+    assert method == "DELETE"
+    assert path.endswith("/api-keys/flux-key-1/bearer-token/")
+    assert not path.endswith("/api-keys/flux-key-1/")
+
+
+def test_flux_api_key_bearer_fields_are_optional():
+    """A server predating the feature omits them; the model must still validate."""
+    from foxnose_sdk import FluxAPIKeySummary
+
+    key = FluxAPIKeySummary.model_validate(FLUX_API_KEY_JSON)
+    assert key.bearer_token_prefix is None
+    assert key.bearer_token_issued_at is None
+
+    with_token = FluxAPIKeySummary.model_validate(
+        FLUX_API_KEY_JSON
+        | {
+            "bearer_token_prefix": "fxk_A7fQ2mXe",
+            "bearer_token_issued_at": "2026-08-09T10:24:11.482Z",
+        }
+    )
+    assert with_token.bearer_token_prefix == "fxk_A7fQ2mXe"
+    # Only the prefix is ever exposed on a key read — never the credential.
+    assert not hasattr(with_token, "bearer_token")
+
+
 def test_management_role_crud():
     captured: list[str] = []