Google Vertex

stream = client.chat.completions.create(
    model="vertex-main/gemini-2.5-flash",
    messages=[{"role": "user", "content": "Count from 1 to 5, one number per line."}],
    stream=True,
)
for chunk in stream:
    if (
        chunk.choices
        and len(chunk.choices) > 0
        and chunk.choices[0].delta.content is not None
    ):
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

import json

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get the current weather for a city.",
        "parameters": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
}]

messages = [{"role": "user", "content": "What's the weather in Bengaluru right now?"}]
first = client.chat.completions.create(
    model="vertex-main/gemini-2.5-flash",
    messages=messages,
    tools=tools,
    tool_choice={"type": "function", "function": {"name": "get_weather"}},
)

assistant_msg = first.choices[0].message
tool_calls = assistant_msg.tool_calls or []
if tool_calls:
    tool_call = tool_calls[0]
    messages.append(assistant_msg)
    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps({"city": "Bengaluru", "temp_c": 28, "summary": "partly cloudy"}),
    })

    second = client.chat.completions.create(
        model="vertex-main/gemini-2.5-flash",
        messages=messages,
    )
    print(second.choices[0].message.content)

import base64
from io import BytesIO
from PIL import Image as PILImage, ImageDraw

img = PILImage.new("RGB", (256, 256), (30, 144, 255))
draw = ImageDraw.Draw(img)
draw.ellipse((48, 48, 208, 208), fill=(255, 215, 0))
draw.rectangle((96, 96, 160, 160), fill=(220, 20, 60))

buf = BytesIO()
img.save(buf, format="PNG")
data_uri = f"data:image/png;base64,{base64.b64encode(buf.getvalue()).decode('ascii')}"

response = client.chat.completions.create(
    model="vertex-main/gemini-3-pro-image-preview",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Describe this image in one sentence."},
            {"type": "image_url", "image_url": {"url": data_uri, "detail": "low"}},
        ],
    }],
)
print(response.choices[0].message.content)

import base64, wave, struct, math
from io import BytesIO

# Generate a 1-second 440 Hz sine wave PCM WAV in-memory
buf = BytesIO()
with wave.open(buf, "wb") as w:
    w.setnchannels(1)
    w.setsampwidth(2)
    w.setframerate(16000)
    for i in range(16000):
        sample = int(32767 * 0.3 * math.sin(2 * math.pi * 440 * i / 16000))
        w.writeframes(struct.pack("<h", sample))
audio_uri = f"data:audio/wav;base64,{base64.b64encode(buf.getvalue()).decode('ascii')}"

response = client.chat.completions.create(
    model="vertex-main/gemini-2.5-flash",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Describe this audio in one sentence."},
            {"type": "image_url", "image_url": {"url": audio_uri, "mime_type": "audio/wav"}},
        ],
    }],
)
print(response.choices[0].message.content)

SAMPLE_VIDEO_URL = "https://www.youtube.com/watch?v=8FsHo7xoTr4"

response = client.chat.completions.create(
    model="vertex-main/gemini-2.5-flash",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "Describe what happens in this video in one sentence."},
            {"type": "image_url", "image_url": {"url": SAMPLE_VIDEO_URL, "mime_type": "video/mp4"}},
        ],
    }],
)
print(response.choices[0].message.content)

import base64

with open("sample.pdf", "rb") as f:
    pdf_b64 = base64.b64encode(f.read()).decode("ascii")

response = client.chat.completions.create(
    model="vertex-main/gemini-2.5-flash",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "What text is in this PDF?"},
            {
                "type": "file",
                "file": {
                    "filename": "sample.pdf",
                    "file_data": f"data:application/pdf;base64,{pdf_b64}",
                },
            },
        ],
    }],
)
print(response.choices[0].message.content)

import json

schema = {
    "name": "person",
    "schema": {
        "type": "object",
        "properties": {
            "name": {"type": "string"},
            "age": {"type": "integer"},
            "hobbies": {"type": "array", "items": {"type": "string"}},
        },
        "required": ["name", "age", "hobbies"],
        "additionalProperties": False,
    },
    "strict": True,
}

response = client.chat.completions.create(
    model="vertex-main/gemini-2.5-flash",
    messages=[{"role": "user", "content": "Invent a fictional person with name, age, and three hobbies."}],
    response_format={"type": "json_schema", "json_schema": schema},
)

message = response.choices[0].message
if getattr(message, "refusal", None):
    print("model refused:", message.refusal)
elif message.content:
    print(json.dumps(json.loads(message.content), indent=2))

response = client.chat.completions.create(
    model="vertex-main/gemini-2.5-pro",
    messages=[{
        "role": "user",
        "content": "A bat and a ball cost $1.10. The bat costs $1.00 more than the ball. How much is the ball?",
    }],
    reasoning_effort="high",
    max_tokens=8000,
)

msg = response.choices[0].message
print("answer:", msg.content)
print("reasoning:", getattr(msg, "reasoning_content", None))

for block in getattr(msg, "thinking_blocks", []) or []:
    print("  block:", block.get("type"), "signature:", block.get("signature", "")[:30])

from openai import OpenAI

batch_client = OpenAI(
    api_key="your-truefoundry-api-key",
    base_url="{GATEWAY_BASE_URL}",
    default_headers={
        "x-tfy-provider-name": "vertex-main",
        "x-tfy-vertex-storage-bucket-name": "your-gcs-bucket-name",
        "x-tfy-vertex-region": "us-central1",
        "x-tfy-provider-model": "gemini-2.5-flash",  # bare Vertex id
    },
)

import json

LANGUAGES = ["French", "Japanese", "Hindi", "Spanish", "German"]
batch_requests = [
    {
        "custom_id": f"req-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gemini-2.5-flash",  # bare Vertex id inside the body too
            "messages": [{"role": "user", "content": f"Say hello in {lang}."}],
            "max_tokens": 50,
        },
    }
    for i, lang in enumerate(LANGUAGES)
]

with open("batch_input.jsonl", "w") as f:
    for req in batch_requests:
        f.write(json.dumps(req) + "\n")

with open("batch_input.jsonl", "rb") as f:
    uploaded = batch_client.files.create(file=f, purpose="batch")
print(uploaded.id)  # Example: gs%3A%2F%2Fyour-bucket%2Fuuid.jsonl (URL-encoded)

batch = batch_client.batches.create(
    input_file_id=uploaded.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)
print("batch id:", batch.id, "status:", batch.status)

import time
from urllib.parse import unquote

TERMINAL = {"completed", "failed", "expired", "cancelled"}
TIMEOUT_SECONDS = 30 * 60
POLL_INTERVAL = 15

batch_id = unquote(batch.id)

start = time.monotonic()
while batch.status not in TERMINAL:
    if time.monotonic() - start > TIMEOUT_SECONDS:
        print(f"timed out after {TIMEOUT_SECONDS}s — rerun this cell to keep polling")
        break
    time.sleep(POLL_INTERVAL)
    batch = batch_client.batches.retrieve(batch_id)
    print("status:", batch.status)

print("final:", batch.status, "output_file_id:", batch.output_file_id)

import json
from urllib.parse import unquote

if batch.status == "completed":
    output_id = unquote(batch.output_file_id)
    text = batch_client.files.content(output_id).read().decode("utf-8")
    rows = json.loads(text)
    print(rows)