feat: accept an OAuth bearer token, not only an API key

README.md

@@ -131,6 +131,30 @@ kw.cores()   # list the account's cores (the default core is included)
 
 `test` and `client1` never share memory. Omit `core` for the account's default brain.
 
+## Two ways to authenticate
+
+An **API key** is a long-lived account secret. It is the right credential when
+the process belongs to you — a script, a job, a server you run:
+
+```python
+Khwan(api_key="kwk_live_…")          # sent as X-API-Key
+```
+
+A **bearer token** is an OAuth access token minted for one end user, short-lived
+and scoped to a resource. It is the right credential when you are acting on
+someone's behalf and should never hold their key — a remote MCP server, or any
+service where the caller authenticated with Khwan rather than with you:
+
+```python
+Khwan(bearer_token=access_token)     # sent as Authorization: Bearer
+```
+
+Pass exactly one. They are not interchangeable at the wire: a token placed in
+`api_key` is looked up as an API key, misses, and 401s — it never reaches the
+bearer path, and the error does not say why.
+
+`core` and `user_id` work the same with either.
+
 ## On-prem
 Same code, point at your instance:
 ```python

src/khwan/__init__.py

@@ -143,8 +143,18 @@ def _error_message(status: int, text: str) -> str:
     }.get(status, text[:300])
 
 
-def _auth_headers(api_key: str, user_id: Optional[str], core: Optional[str]) -> Dict[str, str]:
-    h = {"X-API-Key": api_key}
+def _auth_headers(api_key: Optional[str], user_id: Optional[str], core: Optional[str],
+                  bearer_token: Optional[str] = None) -> Dict[str, str]:
+    """Whichever credential this client was built with.
+
+    Two kinds, and they are not interchangeable at the wire. An API key is a
+    long-lived secret the account owns and sends as `X-API-Key`. A bearer token
+    is an OAuth access token minted for one user, short-lived and scoped to a
+    resource, and the API reads it from `Authorization`. Sending a JWT in the
+    API-key header does not "also work" — it is looked up as a key, fails, and
+    401s before the bearer path is ever reached.
+    """
+    h = {"Authorization": f"Bearer {bearer_token}"} if bearer_token else {"X-API-Key": api_key or ""}
     if user_id:
         h["X-Khwan-User"] = user_id   # optional: isolated sub-brain per end-user
     if core:
@@ -154,6 +164,7 @@ def _auth_headers(api_key: str, user_id: Optional[str], core: Optional[str]) ->
 
 class Khwan:
     def __init__(self, *, user_id: Optional[str] = None, api_key: Optional[str] = None,
+                 bearer_token: Optional[str] = None,
                  base_url: str = DEFAULT_BASE_URL,
                  model: Optional[str] = None, constitution: Optional[str] = None,
                  core: Optional[str] = None,
@@ -164,8 +175,11 @@ def __init__(self, *, user_id: Optional[str] = None, api_key: Optional[str] = No
                 "memory/embedder are server-managed in the hosted client; they are "
                 "only configurable in the on-prem engine (khwan-engine, under license)."
             )
-        if not api_key:
-            raise ValueError("api_key is required (get one from your Khwan dashboard).")
+        if bool(api_key) == bool(bearer_token):
+            raise ValueError(
+                "pass exactly one of api_key or bearer_token — an api_key from your "
+                "Khwan dashboard, or an OAuth access token for one user."
+            )
         # OPTIONAL end-user id. Omit for one shared brain per account/core. Set it to give
         # each of your end-users a fully ISOLATED sub-brain (one key → a private brain per
         # user); requires a paid plan. Combines with `core`: account::<core>::@<user>.
@@ -175,6 +189,7 @@ def __init__(self, *, user_id: Optional[str] = None, api_key: Optional[str] = No
         # Omit for the account's default core.
         self.core = core
         self._key = api_key
+        self._bearer = bearer_token
         self._base = base_url.rstrip("/")
         self._timeout = timeout
         # Auto-retry transient failures (429/502/503/504 honoring Retry-After, plus
@@ -195,7 +210,7 @@ def __init__(self, *, user_id: Optional[str] = None, api_key: Optional[str] = No
 
     # ---- transport ----
     def _headers(self) -> Dict[str, str]:
-        return _auth_headers(self._key, self.user_id, self.core)
+        return _auth_headers(self._key, self.user_id, self.core, self._bearer)
 
     def _request(self, method: str, path: str, body: Optional[dict] = None) -> dict:
         idempotent = _is_idempotent(method, path)
@@ -447,6 +462,7 @@ class AsyncKhwan:
 
     def __init__(self, *, user_id: Optional[str] = None, api_key: Optional[str] = None,
                  base_url: str = DEFAULT_BASE_URL,
+                 bearer_token: Optional[str] = None,
                  model: Optional[str] = None, constitution: Optional[str] = None,
                  core: Optional[str] = None,
                  timeout: int = 60, max_retries: int = 2):
@@ -456,11 +472,15 @@ def __init__(self, *, user_id: Optional[str] = None, api_key: Optional[str] = No
             raise ModuleNotFoundError(
                 'AsyncKhwan needs httpx — install with: pip install "khwan[async]"'
             ) from e
-        if not api_key:
-            raise ValueError("api_key is required (get one from your Khwan dashboard).")
+        if bool(api_key) == bool(bearer_token):
+            raise ValueError(
+                "pass exactly one of api_key or bearer_token — an api_key from your "
+                "Khwan dashboard, or an OAuth access token for one user."
+            )
         self.user_id = user_id
         self.core = core
         self._key = api_key
+        self._bearer = bearer_token
         self._base = base_url.rstrip("/")
         self._timeout = timeout
         self._max_retries = max(0, max_retries)
@@ -483,7 +503,7 @@ def _http(self):
             self._client = httpx.AsyncClient(
                 base_url=self._base,
                 timeout=httpx.Timeout(self._timeout, connect=10.0),
-                headers=_auth_headers(self._key, self.user_id, self.core),
+                headers=_auth_headers(self._key, self.user_id, self.core, self._bearer),
             )
         return self._client
 

test_bearer.py

@@ -0,0 +1,93 @@
+"""Two credential kinds, and they are not interchangeable at the wire.
+
+An API key is a long-lived account secret sent as `X-API-Key`. A bearer token is
+an OAuth access token minted for one user, and the API reads it from
+`Authorization`. A JWT put in the api_key slot does not quietly also work — it is
+looked up as a key, misses, and 401s before the bearer path is reached. These
+tests pin the header each one produces, because that difference is invisible
+until a request fails somewhere else.
+
+Offline: constructing a client makes no network call, and the assertions read the
+headers the client would send.
+"""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent / "src"))
+
+from khwan import Khwan, _auth_headers  # noqa: E402
+
+JWT = "eyJhbGciOiJSUzI1NiJ9.stub.stub"
+
+
+# ── which header goes out ─────────────────────────────────────────────────────
+
+def test_api_key_goes_in_the_api_key_header():
+    h = Khwan(api_key="kwk_live_x")._headers()
+    assert h["X-API-Key"] == "kwk_live_x"
+    assert "Authorization" not in h
+
+
+def test_bearer_goes_in_authorization():
+    h = Khwan(bearer_token=JWT)._headers()
+    assert h["Authorization"] == f"Bearer {JWT}"
+    assert "X-API-Key" not in h
+
+
+def test_a_bearer_client_never_sends_an_api_key_header():
+    """The whole point: X-API-Key resolves as a key and 401s before bearer runs."""
+    assert "X-API-Key" not in _auth_headers(None, None, None, JWT)
+
+
+# ── brain selection is orthogonal to the credential ───────────────────────────
+
+def test_core_and_user_ride_along_with_a_bearer():
+    h = Khwan(bearer_token=JWT, core="acme", user_id="web")._headers()
+    assert h["Authorization"] == f"Bearer {JWT}"
+    assert h["X-Khwan-Core"] == "acme"
+    assert h["X-Khwan-User"] == "web"
+
+
+def test_core_and_user_still_ride_along_with_a_key():
+    h = Khwan(api_key="kwk_live_x", core="acme", user_id="web")._headers()
+    assert (h["X-API-Key"], h["X-Khwan-Core"], h["X-Khwan-User"]) == (
+        "kwk_live_x", "acme", "web")
+
+
+# ── exactly one credential ────────────────────────────────────────────────────
+
+def test_neither_is_refused():
+    with pytest.raises(ValueError, match="exactly one"):
+        Khwan()
+
+
+def test_both_is_refused():
+    """Ambiguous rather than harmless — one of them would silently win."""
+    with pytest.raises(ValueError, match="exactly one"):
+        Khwan(api_key="kwk_live_x", bearer_token=JWT)
+
+
+def test_an_empty_string_is_not_a_credential():
+    with pytest.raises(ValueError, match="exactly one"):
+        Khwan(bearer_token="")
+
+
+# ── the async client agrees ───────────────────────────────────────────────────
+
+def test_async_client_takes_a_bearer_too():
+    httpx = pytest.importorskip("httpx")  # noqa: F841
+    from khwan import AsyncKhwan
+    client = AsyncKhwan(bearer_token=JWT, core="acme")
+    assert client._bearer == JWT
+    assert _auth_headers(client._key, client.user_id, client.core,
+                         client._bearer)["Authorization"] == f"Bearer {JWT}"
+
+
+def test_async_client_refuses_both():
+    pytest.importorskip("httpx")
+    from khwan import AsyncKhwan
+    with pytest.raises(ValueError, match="exactly one"):
+        AsyncKhwan(api_key="kwk_live_x", bearer_token=JWT)