feat(auth): validate credential before the first request

Raise TypeError from _validate_headers when neither pat, credential, nor
QODER_PAT resolves an Authorization header, rather than sending a
gateway-bypass request. A configured dynamic credential is allowed since it
injects auth later in the send pipeline. Mirrors anthropic's request-time
validation and the Go/TS SDKs.

Author: moonyue-w

src/qca/common/_base_client.py

@@ -118,6 +118,17 @@ def _raw_response_view(self, *, streaming: bool = False) -> Self:
     def is_closed(self) -> bool:
         return self._client.is_closed
 
+    def _validate_headers(self, headers: httpx.Headers) -> None:
+        if headers.get("Authorization"):
+            return
+        if self.credential is not None:
+            return
+
+        raise TypeError(
+            "Could not resolve authentication method. Expected one of pat or credential to be set, "
+            "or the QODER_PAT environment variable to be configured."
+        )
+
     def _headers(self, options: dict[str, Any], token: str | None) -> httpx.Headers:
         headers = httpx.Headers(
             {
@@ -132,6 +143,7 @@ def _headers(self, options: dict[str, Any], token: str | None) -> httpx.Headers:
         for source in (self.default_headers, options.get("headers", {})):
             for key, value in source.items():
                 headers[key] = ",".join(str(v) for v in value) if isinstance(value, (list, tuple)) else str(value)
+        self._validate_headers(headers)
         return headers
 
     def _request_args(

tests/test_client.py

@@ -42,6 +42,18 @@ def make_client(handler, cls=Forward, **kwargs):
     )
 
 
+@pytest.mark.parametrize("cls,resource", [(Forward, "templates"), (Managed, "agents")])
+def test_missing_credential_raises_on_first_request(monkeypatch, cls, resource):
+    monkeypatch.delenv("QODER_PAT", raising=False)
+    # Construction succeeds; the error surfaces when we try to build the request.
+    with cls(http_client=httpx.Client(transport=httpx.MockTransport(lambda _: httpx.Response(200, json={"data": [], "has_more": False})))) as client:
+        with pytest.raises(TypeError, match="Could not resolve authentication method"):
+            getattr(client, resource).list()
+    # An explicit PAT lets the same call go through.
+    with cls(pat="explicit", http_client=httpx.Client(transport=httpx.MockTransport(lambda _: httpx.Response(200, json={"data": [], "has_more": False})))) as client:
+        getattr(client, resource).list()
+
+
 @pytest.mark.parametrize(
     "cls,env,suffix", [(Forward, "QODER_FORWARD_BASE_URL", "forward"), (Managed, "QODER_MANAGED_BASE_URL", "cloud")]
 )
@@ -416,7 +428,7 @@ async def handle(request):
         await asyncio.wait_for(both.wait(), timeout=1)
         return httpx.Response(200, json={"id": "identity", "data": []})
 
-    async with AsyncForward(http_client=httpx.AsyncClient(transport=httpx.MockTransport(handle))) as client:
+    async with AsyncForward(pat="test", http_client=httpx.AsyncClient(transport=httpx.MockTransport(handle))) as client:
         first, second = await asyncio.gather(client.identities.retrieve("first"), client.identities.retrieve("second"))
         assert first.id == second.id == "identity"
         raw = await client.identities.with_raw_response.retrieve("third")
@@ -483,6 +495,7 @@ async def aclose(self):
 
     body = Body()
     async with AsyncForward(
+        pat="test",
         http_client=httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=body)))
     ) as client:
         async with client.models.with_streaming_response.list() as response:

tests/test_pagination.py

@@ -19,7 +19,7 @@ def handle(request):
             200, json={"data": [{"id": str(index)}], "has_more": index == 1, "first_id": "first", "last_id": "last"}
         )
 
-    with Forward(http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
+    with Forward(pat="test", http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
         args = {"extra_query": {"name": "filter"}, **({"before_id": "start"} if backward else {})}
         page = client.templates.list(**args)
         assert [item.id for item in page] == ["1", "2"]
@@ -38,13 +38,14 @@ def handle(request):
         return httpx.Response(200, json={"data": [{"id": "two"}], "has_more": False})
 
     with Managed(
-        default_query={"after_id": "old"}, http_client=httpx.Client(transport=httpx.MockTransport(handle))
+        pat="test", default_query={"after_id": "old"}, http_client=httpx.Client(transport=httpx.MockTransport(handle))
     ) as client:
         assert [item.id for item in client.agents.list()] == ["two"]
 
 
 def test_repeated_pagination_cursor_raises_instead_of_looping():
     with Managed(
+        pat="test",
         http_client=httpx.Client(
             transport=httpx.MockTransport(
                 lambda _: httpx.Response(200, json={"data": [{"id": "one"}], "next_page": "same", "has_more": True})
@@ -72,7 +73,7 @@ def handle(request):
             },
         )
 
-    async with cls(http_client=httpx.AsyncClient(transport=httpx.MockTransport(handle))) as client:
+    async with cls(pat="test", http_client=httpx.AsyncClient(transport=httpx.MockTransport(handle))) as client:
         resource = client.agents if cls is AsyncManaged else client.templates
         page = await resource.list()
         assert page.data[0].id == "one"
@@ -87,7 +88,7 @@ def handle(request):
             200, json={"data": [{"id": "one" if more else "two"}], "last_id": "one", "has_more": more}
         )
 
-    with Forward(http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
+    with Forward(pat="test", http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
         page = client.templates.with_raw_response.list().parse()
         assert [item.id for item in page] == ["one", "two"]
 
@@ -98,7 +99,7 @@ def test_cursor_returning_to_an_earlier_value_raises_instead_of_looping():
     def handle(request):
         return httpx.Response(200, json={"data": [{"id": "item"}], "next_page": cursors.pop(0), "has_more": True})
 
-    with Managed(http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
+    with Managed(pat="test", http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
         with pytest.raises(RuntimeError, match="cycle detected"):
             list(client.agents.list())
 
@@ -110,7 +111,7 @@ def handle(request):
             200, json={"data": [{"id": "one" if first else "two"}], "next_page": "next" if first else None}
         )
 
-    with Managed(http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
+    with Managed(pat="test", http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
         page = client.agents.list()
         assert page.has_next_page()
         last = page.get_next_page()

tests/test_streaming.py

@@ -56,7 +56,7 @@ def handle(request):
         assert request.url.params.get_list("event_deltas[]") == ["agent.message"]
         return httpx.Response(200, stream=chunks, headers={"content-type": "text/event-stream"})
 
-    with Forward(http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
+    with Forward(pat="test", http_client=httpx.Client(transport=httpx.MockTransport(handle))) as client:
         with client.sessions.events.stream(
             "session", last_event_id="previous", event_deltas=["agent.message"]
         ) as stream:
@@ -78,6 +78,7 @@ def handle(request):
 def test_stream_errors_close_response(body, error):
     chunks = Chunks(body.encode())
     with Forward(
+        pat="test",
         http_client=httpx.Client(transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=chunks)))
     ) as client:
         with pytest.raises(error):
@@ -98,6 +99,7 @@ async def aclose(self):
 
     chunks = AsyncChunks()
     async with AsyncManaged(
+        pat="test",
         http_client=httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=chunks)))
     ) as client:
         async with await client.sessions.events.stream("session", extra_headers={"Last-Event-ID": "before"}) as stream: