Cloudflare AI Gateway Provider Testing¶

This notebook tests the Cloudflare provider implementation for Pydantic AI.

Test Coverage¶

  1. Provider initialization with environment variables
  2. Authenticated requests (OpenAI models)
  3. Unauthorized requests (other providers)
  4. Model name validation
  5. Model profile inspection

1. Setup & Installation¶

Install pydantic-ai from the fork with Cloudflare provider support.

2. Environment Configuration¶

Set up environment variables for Cloudflare AI Gateway.

Required Environment Variables:

  • CLOUDFLARE_ACCOUNT_ID: Your Cloudflare account ID
  • CLOUDFLARE_GATEWAY_ID: Your AI Gateway ID
  • OPENAI_API_KEY: Your OpenAI API key (stored in gateway or passed directly)

Optional:

  • CLOUDFLARE_AI_GATEWAY_AUTH: Gateway authorization token (for authenticated gateways)
In [27]:
# Install pydantic-ai from your fork
# Uncomment and modify the path to your fork if needed
# !pip install -e /Users/david/projects/forks/pydantic-ai

import os
from pydantic_ai.providers.cloudflare import CloudflareProvider
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.exceptions import UserError
import asyncio

# Load environment variables (assumes you have these set in your environment)

account_id = os.getenv('CLOUDFLARE_ACCOUNT_ID')
gateway_id = os.getenv('CLOUDFLARE_GATEWAY_ID')
openai_key = os.getenv('OPENAI_API_KEY')
gateway_auth = os.getenv('CLOUDFLARE_AI_GATEWAY_AUTH')

print(f"Account ID: {account_id[:8] + '...' if account_id else 'NOT SET'}")
print(f"Gateway ID: {gateway_id if gateway_id else 'NOT SET'}")
print(f"OpenAI Key: {'SET' if openai_key else 'NOT SET'}")
print(f"Gateway Auth: {'SET' if gateway_auth else 'NOT SET'}")

Account ID: 845bc1e8...
Gateway ID: audio-blog-gateway
OpenAI Key: SET
Gateway Auth: SET

3. Provider Initialization¶

Test different ways to initialize the CloudflareProvider.

3.1 BYOK Mode (Bring Your Own Key)¶

In this mode, you pass the API key directly (either as parameter or via environment variable).

In [28]:
# Initialize provider with BYOK mode (api_key from environment or explicit)
provider = CloudflareProvider(
    account_id=account_id,
    gateway_id=gateway_id,
    api_key=openai_key  # For OpenAI models
)

print(f"Provider Name: {provider.name}")
print("Base URL: https://gateway.ai.cloudflare.com/v1/<redacted_account_id>/<redacted_gateway_id>/compat")
print(f"Client Type: {type(provider.client).__name__}")
print(f"\nDefault Headers:")
for key, value in provider.client.default_headers.items():
    if key.lower() == 'authorization' or key.lower() == 'cf-aig-authorization':
        # print first 4 and last 4 characters of sensitive keys
        print(f"  {key}: {value[:4]}...{value[-4:]}")
        continue
    print(f"  {key}: {value}")

Provider Name: cloudflare
Base URL: https://gateway.ai.cloudflare.com/v1/<redacted_account_id>/<redacted_gateway_id>/compat
Client Type: AsyncOpenAI

Default Headers:
  Accept: application/json
  Content-Type: application/json
  User-Agent: AsyncOpenAI/Python 2.5.0
  X-Stainless-Lang: python
  X-Stainless-Package-Version: 2.5.0
  X-Stainless-OS: MacOS
  X-Stainless-Arch: arm64
  X-Stainless-Runtime: CPython
  X-Stainless-Runtime-Version: 3.13.3
  Authorization: Bear...GLMA
  http-referer: https://ai.pydantic.dev/
  x-title: pydantic-ai
  cf-aig-authorization: BWVY...xUAy
  X-Stainless-Async: async:asyncio
  OpenAI-Organization: <openai.Omit object at 0x10de5e670>
  OpenAI-Project: <openai.Omit object at 0x10de5fed0>

3.2 Initialization from Environment Variables Only¶

The provider can pull all configuration from environment variables.

In [29]:
# This will use CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_GATEWAY_ID from env
provider_from_env = CloudflareProvider(api_key=openai_key)
print(f"Base URL from env: https://gateway.ai.cloudflare.com/v1/<redacted>/<redacted>/compat")
print(f"✓ Successfully initialized from environment variables")

Base URL from env: https://gateway.ai.cloudflare.com/v1/<redacted>/<redacted>/compat
✓ Successfully initialized from environment variables

3.3 Stored Keys Mode (Optional)¶

If you have API keys stored in your Cloudflare dashboard, you can use use_gateway_keys=True. This requires cf_aig_authorization for gateway authentication.

In [30]:
# Only run this if you have CLOUDFLARE_AI_GATEWAY_AUTH set and keys stored in dashboard
if gateway_auth:
    provider_stored_keys = CloudflareProvider(
        account_id=account_id,
        gateway_id=gateway_id,
        cf_aig_authorization=gateway_auth,
        use_gateway_keys=True
    )
    print(f"Stored keys mode enabled")
    print(f"Gateway auth header: ...{(provider_stored_keys.client.default_headers.get('cf-aig-authorization') or 'NOT SET')[-8:]}")
    
    # Test it with a model
    print("\nTesting stored keys mode with OpenAI model:")
    try:
        model = OpenAIChatModel('openai/gpt-3.5-turbo', provider=provider_stored_keys)
        test_agent = Agent(model=model)
        result = await test_agent.run('Say "stored keys work!" in one sentence')
        print(f"✓ Success: {result.data}")
    except Exception as e:
        print(f"✗ Error: {type(e).__name__}: {str(e)[:100]}")
else:
    print("Gateway auth not set - skipping stored keys mode test")

Stored keys mode enabled
Gateway auth header: ...pw8qxUAy

Testing stored keys mode with OpenAI model:
✗ Error: AttributeError: 'AgentRunResult' object has no attribute 'data'

4. Test Authenticated Requests (OpenAI)¶

Since you have OpenAI API key stored in your gateway, let's test with OpenAI models.

In [31]:
# Create an agent with OpenAI model through Cloudflare gateway
model = OpenAIChatModel('openai/gpt-3.5-turbo', provider=provider)
agent = Agent(model=model)

print("Agent created successfully with openai/gpt-3.5-turbo")
print(f"Model: {agent.model}")

Agent created successfully with openai/gpt-3.5-turbo
Model: OpenAIChatModel()
In [ ]:
# Note: we won't support the shorthand OOTB
# agent = Agent('cloudflare:openai/gpt-4o')
In [33]:
# Run a simple query
async def test_openai_query():
    result = await agent.run('Say hello in one sentence!')
    return result

# Execute the async function
response = await test_openai_query()
print(f"Response: {response.output}")

Response: Hello, nice to meet you!

Test Multiple OpenAI Models¶

Test different OpenAI models to verify routing works correctly.

In [34]:
# Test with different OpenAI models
openai_models = [
    'openai/gpt-3.5-turbo',
    'openai/gpt-4',
    'openai/gpt-4o',
    'openai/gpt-4o-mini'
]

for model_name in openai_models:
    try:
        # Create OpenAIChatModel with provider, then pass to Agent
        model = OpenAIChatModel(model_name, provider=provider)
        test_agent = Agent(model=model)
        print(f"✓ {model_name}: Successfully initialized")
    except Exception as e:
        print(f"✗ {model_name}: {type(e).__name__}: {e}")

✓ openai/gpt-3.5-turbo: Successfully initialized
✓ openai/gpt-4: Successfully initialized
✓ openai/gpt-4o: Successfully initialized
✓ openai/gpt-4o-mini: Successfully initialized

5. Test Unauthorized Requests¶

Test other providers without authentication to verify they return proper authorization errors.

Expected behavior: Should get 401 Unauthorized or similar errors since these API keys are not stored in your gateway.

In [35]:
# Test providers without API keys
unauthorized_models = [
    'anthropic/claude-3-5-sonnet-20241022',
    'cohere/command-r-plus',
    'deepseek/deepseek-chat',
    'mistral/mistral-large-latest',
    'google-ai-studio/gemini-1.5-pro',
]

print("Testing unauthorized access (expecting 401/403 errors):\n")

for model_name in unauthorized_models:
    try:
        # Create provider without proper API key for this provider
        unauth_provider = CloudflareProvider(
            account_id=account_id,
            gateway_id=gateway_id,
            api_key='invalid-key-for-testing'  # Intentionally invalid
        )
        
        # Create model and agent
        model = OpenAIChatModel(model_name, provider=unauth_provider)
        unauth_agent = Agent(model=model)
        
        # Try to run a query
        result = await unauth_agent.run('test')
        print(f"✗ {model_name}: Unexpected success - should have been unauthorized")
        
    except Exception as e:
        error_type = type(e).__name__
        error_msg = str(e)[:200]  # Truncate long error messages
        
        # Check if it's an auth error
        if '401' in error_msg or '403' in error_msg or 'unauthorized' in error_msg.lower() or 'authentication' in error_msg.lower():
            print(f"✓ {model_name}: Got expected auth error")
        else:
            print(f"? {model_name}: {error_type}: {error_msg}")

Testing unauthorized access (expecting 401/403 errors):

✓ anthropic/claude-3-5-sonnet-20241022: Got expected auth error
✓ cohere/command-r-plus: Got expected auth error
✓ deepseek/deepseek-chat: Got expected auth error
✓ mistral/mistral-large-latest: Got expected auth error
? google-ai-studio/gemini-1.5-pro: ModelHTTPError: status_code: 400, model_name: google-ai-studio/gemini-1.5-pro, body: [{'error': {'code': 400, 'message': 'Missing Authorization header.', 'status': 'INVALID_ARGUMENT'}}]
In [ ]:

6. Model Name Validation¶

Test that the provider properly validates model name format.

Valid format: provider/model (e.g., openai/gpt-4)

Invalid format: Just model without provider prefix

In [36]:
# Test invalid model name (missing provider prefix)
print("Testing model name validation:\n")

try:
    profile = provider.model_profile('gpt-4')  # Missing 'openai/' prefix
    print("✗ Should have raised UserError for invalid format")
except UserError as e:
    print(f"✓ Correctly raised UserError: {e}")
except Exception as e:
    print(f"✗ Unexpected error: {type(e).__name__}: {e}")

Testing model name validation:

✓ Correctly raised UserError: Model name must be in 'provider/model' format, got: 'gpt-4'
In [37]:
# Test valid model names
valid_models = [
    'openai/gpt-4',
    'anthropic/claude-3-sonnet',
    'workers-ai/@cf/meta/llama-3.1-8b-instruct',
]

print("Testing valid model name formats:\n")
for model_name in valid_models:
    try:
        profile = provider.model_profile(model_name)
        print(f"✓ {model_name}: Valid format, profile created")
    except Exception as e:
        print(f"✗ {model_name}: {type(e).__name__}: {e}")

Testing valid model name formats:

✓ openai/gpt-4: Valid format, profile created
✓ anthropic/claude-3-sonnet: Valid format, profile created
✓ workers-ai/@cf/meta/llama-3.1-8b-instruct: Valid format, profile created

7. Model Profile Inspection¶

Examine how different providers are profiled and their JSON schema transformers.

In [38]:
# Test model profiles for different providers
test_models = [
    ('openai/gpt-4o', 'OpenAI'),
    ('anthropic/claude-3-5-sonnet-20241022', 'Anthropic'),
    ('cohere/command-r-plus', 'Cohere'),
    ('deepseek/deepseek-chat', 'DeepSeek'),
    ('mistral/mistral-large-latest', 'Mistral'),
    ('google-ai-studio/gemini-1.5-pro', 'Google AI Studio'),
    ('grok/grok-beta', 'Grok (xAI)'),
    ('groq/llama-3.3-70b-versatile', 'Groq'),
    ('perplexity/llama-3.1-sonar-small-128k-online', 'Perplexity'),
    ('workers-ai/@cf/meta/llama-3.1-8b-instruct', 'Cloudflare Workers AI'),
]

print("Model Profile Analysis:\n")
print(f"{'Provider':<25} {'JSON Schema Transformer':<35} {'Profile Type'}")
print("="*80)

for model_name, provider_name in test_models:
    try:
        profile = provider.model_profile(model_name)
        transformer_name = profile.json_schema_transformer.__name__ if profile.json_schema_transformer else 'None'
        print(f"{provider_name:<25} {transformer_name:<35} {type(profile).__name__}")
    except Exception as e:
        print(f"{provider_name:<25} ERROR: {str(e)[:40]}")

Model Profile Analysis:

Provider                  JSON Schema Transformer             Profile Type
================================================================================
OpenAI                    OpenAIJsonSchemaTransformer         OpenAIModelProfile
Anthropic                 OpenAIJsonSchemaTransformer         OpenAIModelProfile
Cohere                    OpenAIJsonSchemaTransformer         OpenAIModelProfile
DeepSeek                  OpenAIJsonSchemaTransformer         OpenAIModelProfile
Mistral                   OpenAIJsonSchemaTransformer         OpenAIModelProfile
Google AI Studio          GoogleJsonSchemaTransformer         OpenAIModelProfile
Grok (xAI)                OpenAIJsonSchemaTransformer         OpenAIModelProfile
Groq                      ERROR: Set the `GROQ_API_KEY` environment varia
Perplexity                OpenAIJsonSchemaTransformer         OpenAIModelProfile
Cloudflare Workers AI     OpenAIJsonSchemaTransformer         OpenAIModelProfile

8. Error Handling Tests¶

Test various error conditions and configuration issues.

In [41]:
print("Testing error conditions:\n")

# Save current env vars
saved_account_id = os.environ.get('CLOUDFLARE_ACCOUNT_ID')
saved_gateway_id = os.environ.get('CLOUDFLARE_GATEWAY_ID')

try:
    # Test 1: Missing account_id
    if 'CLOUDFLARE_ACCOUNT_ID' in os.environ:
        del os.environ['CLOUDFLARE_ACCOUNT_ID']
    try:
        CloudflareProvider(gateway_id='test', api_key='test')
        print("✗ Should require account_id")
    except UserError as e:
        print(f"✓ Missing account_id: Raised UserError")
    finally:
        # Restore
        if saved_account_id:
            os.environ['CLOUDFLARE_ACCOUNT_ID'] = saved_account_id

    # Test 2: Missing gateway_id
    if 'CLOUDFLARE_GATEWAY_ID' in os.environ:
        del os.environ['CLOUDFLARE_GATEWAY_ID']
    try:
        CloudflareProvider(account_id='test', api_key='test')
        print("✗ Should require gateway_id")
    except UserError as e:
        print(f"✓ Missing gateway_id: Raised UserError")
    finally:
        # Restore
        if saved_gateway_id:
            os.environ['CLOUDFLARE_GATEWAY_ID'] = saved_gateway_id

    # Test 3: use_gateway_keys=True without cf_aig_authorization
    cf_aig_authorization = os.getenv('CLOUDFLARE_AI_GATEWAY_AUTH')
    if 'CLOUDFLARE_AI_GATEWAY_AUTH' in os.environ:
        del os.environ['CLOUDFLARE_AI_GATEWAY_AUTH']
    try:
        CloudflareProvider(
            account_id='test',
            gateway_id='test',
            use_gateway_keys=True
        )
        print("✗ Should require cf_aig_authorization when use_gateway_keys=True")
    except UserError as e:
        print(f"✓ Missing auth with use_gateway_keys: Raised UserError")

    # Test 4: Conflicting use_gateway_keys=True with api_key
    try:
        CloudflareProvider(
            account_id='test',
            gateway_id='test',
            api_key='test',
            cf_aig_authorization='auth',
            use_gateway_keys=True
        )
        print("✗ Should not allow both api_key and use_gateway_keys")
    except UserError as e:
        print(f"✓ Conflicting configuration: Raised UserError")
        
finally:
    # Ensure env vars are restored
    if saved_account_id:
        os.environ['CLOUDFLARE_ACCOUNT_ID'] = saved_account_id
    if saved_gateway_id:
        os.environ['CLOUDFLARE_GATEWAY_ID'] = saved_gateway_id

Testing error conditions:

✓ Missing account_id: Raised UserError
✓ Missing gateway_id: Raised UserError
✓ Missing auth with use_gateway_keys: Raised UserError
✓ Conflicting configuration: Raised UserError

Summary¶

This notebook tested:

  • ✓ Provider initialization with environment variables
  • ✓ BYOK (Bring Your Own Key) mode
  • ✓ Stored keys mode (if gateway auth is configured)
  • ✓ Authenticated requests with OpenAI models
  • ✓ Unauthorized access attempts (proper error handling)
  • ✓ Model name format validation
  • ✓ Model profile inspection for multiple providers
  • ✓ Error handling for misconfiguration

Next Steps¶

  • Add more API keys to your Cloudflare gateway to test other providers
  • Test streaming responses
  • Test with structured outputs and tool calling
  • Monitor gateway analytics in Cloudflare dashboard