From f660cc9bbdd13352101623cd991360bd1aa089f7 Mon Sep 17 00:00:00 2001 From: Constantine Nathanson Date: Wed, 1 Jul 2026 12:52:08 +0300 Subject: [PATCH] Add support for `create_agent_account` Provisioning API Co-Authored-By: Claude Opus 4.8 (1M context) --- cloudinary/api_client/call_account_api.py | 35 +++++++-- cloudinary/api_client/execute_request.py | 10 +-- cloudinary/provisioning/__init__.py | 3 +- cloudinary/provisioning/account.py | 42 +++++++++- test/test_provisioning_api.py | 93 ++++++++++++++++++++++- 5 files changed, 168 insertions(+), 15 deletions(-) diff --git a/cloudinary/api_client/call_account_api.py b/cloudinary/api_client/call_account_api.py index 5a6cf3ab..7e1f4020 100644 --- a/cloudinary/api_client/call_account_api.py +++ b/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, diff --git a/cloudinary/api_client/execute_request.py b/cloudinary/api_client/execute_request.py index 84c155d8..16d0bf45 100644 --- a/cloudinary/api_client/execute_request.py +++ b/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))) diff --git a/cloudinary/provisioning/__init__.py b/cloudinary/provisioning/__init__.py index 7016343a..1f8005c6 100644 --- a/cloudinary/provisioning/__init__.py +++ b/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, diff --git a/cloudinary/provisioning/account.py b/cloudinary/provisioning/account.py index 90b1c385..f54382a0 100644 --- a/cloudinary/provisioning/account.py +++ b/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 diff --git a/test/test_provisioning_api.py b/test/test_provisioning_api.py index 132c1966..80f88f3b 100644 --- a/test/test_provisioning_api.py +++ b/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()