Omit Authorization header when api_key is empty

.fern/metadata.json

@@ -103,7 +103,7 @@
   "originGitCommit": "3f4bc18b3d90a318965a1fc2f5980339c012a6e2",
   "originGitCommitIsDirty": true,
   "invokedBy": "ci",
-  "requestedVersion": "7.0.8",
+  "requestedVersion": "7.0.9",
   "ciProvider": "github",
-  "sdkVersion": "7.0.8"
+  "sdkVersion": "7.0.9"
 }

pyproject.toml

@@ -4,7 +4,7 @@ dynamic = ["version"]
 
 [tool.poetry]
 name = "cohere"
-version = "7.0.8"
+version = "7.0.9"
 description = ""
 readme = "README.md"
 authors = []

src/cohere/core/client_wrapper.py

@@ -35,12 +35,12 @@ def get_headers(self) -> typing.Dict[str, str]:
         import platform
 
         headers: typing.Dict[str, str] = {
-            "User-Agent": "cohere/7.0.8",
+            "User-Agent": "cohere/7.0.9",
             "X-Fern-Language": "Python",
             "X-Fern-Runtime": f"python/{platform.python_version()}",
             "X-Fern-Platform": f"{platform.system().lower()}/{platform.release()}",
             "X-Fern-SDK-Name": "cohere",
-            "X-Fern-SDK-Version": "7.0.8",
+            "X-Fern-SDK-Version": "7.0.9",
             **(self.get_custom_headers() or {}),
         }
         if self._client_name is not None:

src/cohere/overrides.py

@@ -59,12 +59,50 @@ def patched_init(self, /, **data):
     return cls
 
 
+def omit_authorization_header_when_api_key_is_empty() -> None:
+    """
+    Do not send an `Authorization` header when the client is created with an empty API key.
+
+    This allows pointing the client at a proxy or a self-hosted deployment that performs its own
+    authentication, e.g. `cohere.Client(api_key="")`.
+    """
+    from .core.client_wrapper import AsyncClientWrapper, BaseClientWrapper
+
+    if getattr(BaseClientWrapper, "_omits_empty_authorization", False):
+        return
+
+    get_headers = BaseClientWrapper.get_headers
+    async_get_headers = AsyncClientWrapper.async_get_headers
+
+    def patched_get_headers(self: BaseClientWrapper) -> typing.Dict[str, str]:
+        headers = get_headers(self)
+        # Inspect the header that get_headers() already built rather than calling _get_token()
+        # again: for the callable `api_key` form that would invoke the supplier twice per request,
+        # and a supplier whose value changes between the two calls would produce the wrong header.
+        if headers.get("Authorization") == "Bearer ":
+            headers.pop("Authorization", None)
+        return headers
+
+    async def patched_async_get_headers(self: AsyncClientWrapper) -> typing.Dict[str, str]:
+        headers = await async_get_headers(self)
+        if headers.get("Authorization") == "Bearer ":
+            headers.pop("Authorization", None)
+        return headers
+
+    BaseClientWrapper.get_headers = patched_get_headers  # type: ignore[method-assign]
+    AsyncClientWrapper.async_get_headers = patched_async_get_headers  # type: ignore[method-assign]
+    BaseClientWrapper._omits_empty_authorization = True  # type: ignore[attr-defined]
+
+
 def run_overrides():
     """
         These are overrides to allow us to make changes to generated code without touching the generated files themselves.
         Should be used judiciously!
     """
 
+    # Override to skip the Authorization header entirely when an empty api_key is passed
+    omit_authorization_header_when_api_key_is_empty()
+
     # Override to allow access to aliases in EmbedByTypeResponseEmbeddings eg embeddings.float rather than embeddings.float_
     setattr(EmbedByTypeResponseEmbeddings, "__getattr__", allow_access_to_aliases)
 

tests/test_optional_auth.py

@@ -0,0 +1,49 @@
+import asyncio
+import typing
+import unittest
+
+import cohere
+
+
+def _headers(client: typing.Any) -> typing.Dict[str, str]:
+    return client._client_wrapper.get_headers()
+
+
+async def _async_headers(client: typing.Any) -> typing.Dict[str, str]:
+    return await client._client_wrapper.async_get_headers()
+
+
+class TestOptionalAuth(unittest.TestCase):
+    def test_empty_api_key_omits_authorization_header(self) -> None:
+        self.assertNotIn("Authorization", _headers(cohere.Client(api_key="")))
+        self.assertNotIn("Authorization", _headers(cohere.ClientV2(api_key="")))
+        self.assertNotIn("Authorization", asyncio.run(_async_headers(cohere.AsyncClient(api_key=""))))
+        self.assertNotIn("Authorization", asyncio.run(_async_headers(cohere.AsyncClientV2(api_key=""))))
+
+    def test_api_key_is_sent_when_provided(self) -> None:
+        self.assertEqual(_headers(cohere.Client(api_key="n/a"))["Authorization"], "Bearer n/a")
+        self.assertEqual(_headers(cohere.ClientV2(api_key="n/a"))["Authorization"], "Bearer n/a")
+        self.assertEqual(
+            asyncio.run(_async_headers(cohere.AsyncClient(api_key="n/a")))["Authorization"], "Bearer n/a"
+        )
+
+    def test_callable_api_key_returning_empty_string_omits_authorization_header(self) -> None:
+        self.assertNotIn("Authorization", _headers(cohere.Client(api_key=lambda: "")))
+
+    def test_callable_api_key_is_invoked_once_per_request(self) -> None:
+        calls = 0
+
+        def api_key() -> str:
+            nonlocal calls
+            calls += 1
+            return "n/a"
+
+        self.assertEqual(_headers(cohere.Client(api_key=api_key))["Authorization"], "Bearer n/a")
+        self.assertEqual(calls, 1)
+
+    def test_callable_api_key_is_not_re_read_after_the_header_is_built(self) -> None:
+        # A supplier whose value changes between calls must not be able to strip an Authorization
+        # header that was built from a valid token.
+        values = iter(["real-token", ""])
+        client = cohere.Client(api_key=lambda: next(values))
+        self.assertEqual(_headers(client)["Authorization"], "Bearer real-token")