Add support for create_agent_account Provisioning API

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: const-cloudinary

cloudinary/api_client/call_account_api.py

@@ -8,22 +8,43 @@
 _http = get_http_connector(account_config(), cloudinary.CERT_KWARGS)
 
 
+# Account-scoped, authenticated call: provisioning/accounts/{account_id}/...
 def _call_account_api(method, uri, params=None, headers=None, **options):
-    prefix = options.pop("upload_prefix",
-                         cloudinary.config().upload_prefix) or "https://api.cloudinary.com"
+    account_uri = [ACCOUNT_SUB_PATH, _account_id(options)] + uri
+    return _execute_account_request(method, account_uri, _account_auth(options),
+                                    params=params, headers=headers, **options)
+
+
+# Public, unauthenticated call: provisioning/... with no account_id or credentials
+def _call_public_account_api(method, uri, params=None, headers=None, **options):
+    return _execute_account_request(method, uri, {"anonymous": True},
+                                    params=params, headers=headers, **options)
+
+
+def _account_id(options):
     account_id = options.pop("account_id", account_config().account_id)
     if not account_id:
         raise Exception("Must supply account_id")
+    return account_id
+
+
+def _account_auth(options):
     provisioning_api_key = options.pop("provisioning_api_key", account_config().provisioning_api_key)
     if not provisioning_api_key:
         raise Exception("Must supply provisioning_api_key")
-    provisioning_api_secret = options.pop("provisioning_api_secret",
-                                          account_config().provisioning_api_secret)
+    provisioning_api_secret = options.pop("provisioning_api_secret", account_config().provisioning_api_secret)
     if not provisioning_api_secret:
         raise Exception("Must supply provisioning_api_secret")
-    provisioning_api_url = "/".join(
-        [prefix, cloudinary.API_VERSION, PROVISIONING_SUB_PATH, ACCOUNT_SUB_PATH, account_id] + uri)
-    auth = {"key": provisioning_api_key, "secret": provisioning_api_secret}
+    return {"key": provisioning_api_key, "secret": provisioning_api_secret}
+
+
+# Core transport: builds the provisioning URL and dispatches with the resolved auth.
+# The API version can be overridden via the "api_version" option (defaults to cloudinary.API_VERSION).
+def _execute_account_request(method, uri, auth, params=None, headers=None, **options):
+    prefix = options.pop("upload_prefix",
+                         cloudinary.config().upload_prefix) or "https://api.cloudinary.com"
+    api_version = options.pop("api_version", cloudinary.API_VERSION)
+    provisioning_api_url = "/".join([prefix, api_version, PROVISIONING_SUB_PATH] + uri)
 
     return execute_request(http_connector=_http,
                            method=method,

cloudinary/api_client/execute_request.py

@@ -41,14 +41,14 @@ def __init__(self, result, response, **kwargs):
 
 
 def execute_request(http_connector, method, params, headers, auth, api_url, **options):
-    # authentication
+    anonymous = auth.get("anonymous")
     key = auth.get("key")
     secret = auth.get("secret")
     oauth_token = auth.get("oauth_token")
-    req_headers = urllib3.make_headers(
-        user_agent=cloudinary.get_user_agent()
-    )
-    if oauth_token:
+    req_headers = urllib3.make_headers(user_agent=cloudinary.get_user_agent())
+    if anonymous:
+        pass
+    elif oauth_token:
         req_headers["authorization"] = "Bearer {}".format(oauth_token)
     else:
         req_headers.update(urllib3.make_headers(basic_auth="{0}:{1}".format(key, secret)))

cloudinary/provisioning/__init__.py

@@ -1,5 +1,6 @@
 from .account_config import AccountConfig, account_config, reset_config
-from .account import (sub_accounts, create_sub_account, delete_sub_account, sub_account, update_sub_account,
+from .account import (create_agent_account,
+                      sub_accounts, create_sub_account, delete_sub_account, sub_account, update_sub_account,
                       user_groups, create_user_group, update_user_group, delete_user_group, user_group,
                       add_user_to_group, remove_user_from_group, user_group_users, user_in_user_groups,
                       users, create_user, delete_user, user, update_user, access_keys, generate_access_key,

cloudinary/provisioning/account.py

@@ -1,6 +1,7 @@
-from cloudinary.api_client.call_account_api import _call_account_api
+from cloudinary.api_client.call_account_api import _call_account_api, _call_public_account_api, ACCOUNT_SUB_PATH
 from cloudinary.utils import encode_list
 
+AGENTS_SUB_PATH = "agents"
 SUB_ACCOUNTS_SUB_PATH = "sub_accounts"
 USERS_SUB_PATH = "users"
 USER_GROUPS_SUB_PATH = "user_groups"
@@ -20,6 +21,45 @@ class Role(object):
     MEDIA_LIBRARY_USER = "media_library_user"
 
 
+def create_agent_account(email, agent_framework, agent_llm_model, agent_goal, sdk_framework=None, **options):
+    """
+    Create a Cloudinary account on behalf of a human, intended for use by AI agents.
+
+    Creates a Free-plan account with a single, initially disabled product environment and sends a
+    verification email so the human can set a password and activate the account. The returned
+    credentials are inert until the email is verified.
+
+    This endpoint is public and unauthenticated, and is rate-limited per IP address.
+
+    :param email:           The email address of the human on whose behalf the account is created.
+                            A verification email is sent to this address.
+    :type email:            str
+    :param agent_framework: The name of the agent framework used to create the account.
+                            Must be between 2 and 100 characters.
+    :type agent_framework:  str
+    :param agent_llm_model: The LLM model powering the agent. Must be between 2 and 100 characters.
+    :type agent_llm_model:  str
+    :param agent_goal:      A short description of what the agent is trying to achieve.
+                            Must be between 2 and 300 characters.
+    :type agent_goal:       str
+    :param sdk_framework:   The Cloudinary SDK framework the agent intends to use.
+                            Must be between 2 and 100 characters.
+    :type sdk_framework:    str, optional
+    :param options:         Generic advanced options dict, see online documentation
+    :type options:          dict, optional
+    :return:                The created agent account, including inert credentials for its
+                            single product environment
+    :rtype:                 dict
+    """
+    uri = [AGENTS_SUB_PATH, ACCOUNT_SUB_PATH]
+    params = {"email": email,
+              "agent_framework": agent_framework,
+              "agent_llm_model": agent_llm_model,
+              "agent_goal": agent_goal,
+              "sdk_framework": sdk_framework}
+    return _call_public_account_api("POST", uri, params=params, **options)
+
+
 def sub_accounts(enabled=None, ids=None, prefix=None, **options):
     """
     List all sub accounts

test/test_provisioning_api.py

@@ -1,3 +1,4 @@
+import json
 import unittest
 from datetime import datetime
 
@@ -8,7 +9,8 @@
 from cloudinary.provisioning import account_config, reset_config
 from cloudinary.exceptions import AuthorizationRequired, NotFound
 
-from test.helper_test import UNIQUE_SUB_ACCOUNT_ID, UNIQUE_TEST_ID
+from test.helper_test import (UNIQUE_SUB_ACCOUNT_ID, UNIQUE_TEST_ID, URLLIB3_REQUEST, patch, api_response_mock,
+                              get_uri, get_method, get_params, get_headers)
 
 disable_warnings()
 
@@ -85,6 +87,7 @@ def test_update_sub_account(self):
     @unittest.skipUnless(cloudinary.provisioning.account_config().provisioning_api_secret,
                          "requires provisioning_api_key/provisioning_api_secret")
     def test_get_all_sub_accounts(self):
+
         res = cloudinary.provisioning.sub_accounts(True)
 
         sub_account_by_id = [sub_account for sub_account in res["sub_accounts"]
@@ -264,5 +267,93 @@ def test_delete_access_key(self):
         self.assertEqual("ok", named_key_del_res["message"])
 
 
+class CreateAgentAccountTest(unittest.TestCase):
+    """
+    The create agent account endpoint is public, unauthenticated and rate limited per IP,
+    so it is verified against a mocked transport rather than the live API.
+    """
+
+    def test_create_agent_account(self):
+        with patch(URLLIB3_REQUEST) as mocker:
+            mocker.return_value = api_response_mock()
+            cloudinary.provisioning.create_agent_account(
+                "jane@example.com",
+                agent_framework="langchain",
+                agent_llm_model="claude-opus-4-8",
+                agent_goal="Build a product image gallery",
+                sdk_framework="python",
+            )
+
+        self.assertEqual("POST", get_method(mocker))
+        self.assertTrue(get_uri(mocker).endswith("/provisioning/agents/accounts"))
+
+        params = get_params(mocker)
+        self.assertEqual("jane@example.com", params["email"])
+        self.assertEqual("langchain", params["agent_framework"])
+        self.assertEqual("claude-opus-4-8", params["agent_llm_model"])
+        self.assertEqual("Build a product image gallery", params["agent_goal"])
+        self.assertEqual("python", params["sdk_framework"])
+
+    def test_create_agent_account_is_unauthenticated(self):
+        with patch(URLLIB3_REQUEST) as mocker:
+            mocker.return_value = api_response_mock()
+            cloudinary.provisioning.create_agent_account(
+                "jane@example.com",
+                agent_framework="langchain",
+                agent_llm_model="claude-opus-4-8",
+                agent_goal="Build a product image gallery",
+            )
+
+        # The endpoint is public - no authorization header must be sent.
+        headers = get_headers(mocker)
+        self.assertNotIn("authorization", {k.lower() for k in headers})
+
+    def test_create_agent_account_omits_unset_sdk_framework(self):
+        with patch(URLLIB3_REQUEST) as mocker:
+            mocker.return_value = api_response_mock()
+            cloudinary.provisioning.create_agent_account(
+                "jane@example.com",
+                agent_framework="langchain",
+                agent_llm_model="claude-opus-4-8",
+                agent_goal="Build a product image gallery",
+            )
+
+        self.assertNotIn("sdk_framework", get_params(mocker))
+
+    def test_create_agent_account_parses_response(self):
+        body = json.dumps({
+            "external_id": "0aaaaa1bbbbb2ccccc3ddddd4eeeee5f",
+            "email": "jane@example.com",
+            "plan_name": "free",
+            "product_environments": [{
+                "external_id": "abcde1fghij2klmno3pqrst4uvwxy5z",
+                "cloud_name": "product1",
+                "api_key": "123456789012345",
+                "api_secret": "asdf1JKL2xyz3ABc4s3c5reT01DfaKez",
+                "api_environment_variable":
+                    "CLOUDINARY_URL=cloudinary://123456789012345:asdf1JKL2xyz3ABc4s3c5reT01DfaKez@product1",
+            }],
+            "guidance": "A verification email has been sent to the supplied email address.",
+        })
+        with patch(URLLIB3_REQUEST) as mocker:
+            mocker.return_value = api_response_mock(body)
+            res = cloudinary.provisioning.create_agent_account(
+                "jane@example.com",
+                agent_framework="langchain",
+                agent_llm_model="claude-opus-4-8",
+                agent_goal="Build a product image gallery",
+            )
+
+        self.assertEqual("free", res["plan_name"])
+        self.assertEqual("jane@example.com", res["email"])
+        self.assertEqual(1, len(res["product_environments"]))
+        product_environment = res["product_environments"][0]
+        self.assertEqual("product1", product_environment["cloud_name"])
+        self.assertEqual("123456789012345", product_environment["api_key"])
+        self.assertEqual("asdf1JKL2xyz3ABc4s3c5reT01DfaKez", product_environment["api_secret"])
+        self.assertIn("CLOUDINARY_URL=cloudinary://", product_environment["api_environment_variable"])
+        self.assertIn("guidance", res)
+
+
 if __name__ == '__main__':
     unittest.main()