Cloudflare Workers AI API - Edge AI Inference Interface

pip install requests

import requests
import os

# Configuration
CLOUDFLARE_ACCOUNT_ID = "your-account-id"
CLOUDFLARE_API_TOKEN = "your-api-token"

# API endpoint
url = f"https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/ai/run/@cf/meta/llama-3.1-8b-instruct"

# Request headers
headers = {
    "Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}",
    "Content-Type": "application/json"
}

# Request data
data = {
    "messages": [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is Cloudflare Workers AI?"}
    ]
}

# Send request
response = requests.post(url, headers=headers, json=data)
result = response.json()

# Print result
print(result["result"]["response"])

import requests
import json

# Enable streaming response
data = {
    "messages": [
        {"role": "user", "content": "Write a short story about AI"}
    ],
    "stream": True
}

response = requests.post(url, headers=headers, json=data, stream=True)

# Handle streaming data
for line in response.iter_lines():
    if line:
        line = line.decode('utf-8')
        if line.startswith('data: '):
            try:
                data = json.loads(line[6:])
                if 'response' in data:
                    print(data['response'], end='', flush=True)
            except:
                pass

import requests
import base64

# Image generation API
url = f"https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/ai/run/@cf/stabilityai/stable-diffusion-xl-base-1.0"

data = {
    "prompt": "A beautiful sunset over mountains, digital art style"
}

response = requests.post(url, headers=headers, json=data)
result = response.json()

# Save image
image_data = base64.b64decode(result["result"]["image"])
with open("output.png", "wb") as f:
    f.write(image_data)
print("Image saved as output.png")

name = "my-ai-worker"
main = "src/index.js"
compatibility_date = "2024-01-01"

[ai]
binding = "AI"

export default {
  async fetch(request, env) {
    // Use AI binding
    const response = await env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
      messages: [
        { role: "system", content: "You are a helpful assistant." },
        { role: "user", content: "Explain Workers AI in one sentence." }
      ]
    });

    return new Response(JSON.stringify(response), {
      headers: { "content-type": "application/json" }
    });
  }
};

npx wrangler deploy

// Text embeddings
const embeddings = await env.AI.run("@cf/baai/bge-base-en-v1.5", {
  text: "Cloudflare Workers AI is amazing"
});

// Image generation
const imageResponse = await env.AI.run("@cf/stabilityai/stable-diffusion-xl-base-1.0", {
  prompt: "A futuristic city"
});

// Speech-to-text
const transcription = await env.AI.run("@cf/openai/whisper", {
  audio: audioArrayBuffer
});

from openai import OpenAI

# Configure Workers AI
client = OpenAI(
    api_key=CLOUDFLARE_API_TOKEN,
    base_url=f"https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/ai/v1"
)

# Use OpenAI-like API
response = client.chat.completions.create(
    model="@cf/meta/llama-3.1-8b-instruct",
    messages=[
        {"role": "user", "content": "Hello, how are you?"}
    ]
)

print(response.choices[0].message.content)

curl https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/run/@cf/meta/llama-3.1-8b-instruct \
  -H "Authorization: Bearer {api_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "system", "content": "You are a helpful assistant."},
      {"role": "user", "content": "What is edge computing?"}
    ]
  }'

curl https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/run/@cf/meta/llama-3.1-8b-instruct \
  -H "Authorization: Bearer {api_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [{"role": "user", "content": "Tell me a story"}],
    "stream": true
  }'

import time

def call_workers_ai_with_retry(url, headers, data, max_retries=3):
    """API call with retry mechanism"""
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=data, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"Request failed, retrying in {wait_time}s... ({attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                raise Exception(f"Request failed: {e}")

export default {
  async fetch(request, env) {
    const { message } = await request.json();
    
    // Use Llama 3 for fast response
    const response = await env.AI.run(
      "@cf/meta/llama-3.1-8b-instruct",
      {
        messages: [
          {
            role: "system",
            content: "You are a helpful customer service agent."
          },
          { role: "user", content: message }
        ],
        stream: true
      }
    );

    return new Response(response, {
      headers: { "content-type": "text/event-stream" }
    });
  }
};

export default {
  async fetch(request, env) {
    const { query } = await request.json();
    
    // Generate query embeddings
    const embeddings = await env.AI.run(
      "@cf/baai/bge-base-en-v1.5",
      { text: query }
    );
    
    // Use Vectorize for similarity search
    const results = await env.VECTORIZE.query(embeddings.data[0], {
      topK: 5
    });
    
    return Response.json(results);
  }
};

import requests
import base64

def generate_image(prompt, account_id, api_token):
    """Generate image"""
    url = f"https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/run/@cf/bytedance/stable-diffusion-xl-lightning"
    
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Content-Type": "application/json"
    }
    
    data = {
        "prompt": prompt,
        "num_steps": 4  # Lightning model, 4 steps sufficient
    }
    
    response = requests.post(url, headers=headers, json=data)
    result = response.json()
    
    # Save image
    image_data = base64.b64decode(result["result"]["image"])
    filename = f"generated_{hash(prompt)}.png"
    with open(filename, "wb") as f:
        f.write(image_data)
    
    return filename

# Usage example
image_path = generate_image(
    "A futuristic cityscape at sunset, cyberpunk style",
    "your-account-id",
    "your-api-token"
)
print(f"Image saved to: {image_path}")

export default {
  async fetch(request, env) {
    // Receive audio file
    const formData = await request.formData();
    const audioFile = formData.get('audio');
    const audioBuffer = await audioFile.arrayBuffer();
    
    // Use Whisper for transcription
    const transcription = await env.AI.run(
      "@cf/openai/whisper",
      {
        audio: [...new Uint8Array(audioBuffer)]
      }
    );
    
    return Response.json({
      text: transcription.text,
      language: transcription.language
    });
  }
};