Calling models from code

import os

from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential
from dotenv import load_dotenv

load_dotenv()  # reads FOUNDRY_PROJECT_ENDPOINT and FOUNDRY_MODEL_DEPLOYMENT from .env

project = AIProjectClient(
    endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    credential=DefaultAzureCredential(),
)

with project.get_openai_client() as client:
    response = client.responses.create(
        model=os.environ["FOUNDRY_MODEL_DEPLOYMENT"],   # deployment name, e.g. "gpt-5-mini"
        instructions="You are a concise assistant for Azure engineers.",
        input="In one sentence, what is a Foundry project?",
    )

print(response.output_text)
print(response.usage.input_tokens, "in /", response.usage.output_tokens, "out")

import os

from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import OpenAI

token_provider = get_bearer_token_provider(
    DefaultAzureCredential(), "https://ai.azure.com/.default"
)

client = OpenAI(
    base_url=os.environ["FOUNDRY_PROJECT_ENDPOINT"].rstrip("/") + "/openai/v1",
    api_key=token_provider,          # a callable: refreshed automatically, no api-version needed
)

response = client.responses.create(
    model=os.environ["FOUNDRY_MODEL_DEPLOYMENT"],
    input="Give me three uses of Azure AI Search.",
)
print(response.output_text)

# the same question with Chat Completions, on the account-level OpenAI endpoint
account_client = OpenAI(
    base_url=f"https://{resource_name}.openai.azure.com/openai/v1",
    api_key=token_provider,
)
completion = account_client.chat.completions.create(
    model=os.environ["FOUNDRY_MODEL_DEPLOYMENT"],
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "What is a deployment in Foundry?"},
    ],
)
print(completion.choices[0].message.content)
print(completion.choices[0].finish_reason)   # "stop", "length", "content_filter", "tool_calls"

first = client.responses.create(
    model=deployment,
    input="My name is Priya and I manage the support team.",
)
second = client.responses.create(
    model=deployment,
    previous_response_id=first.id,          # the service replays the earlier turn
    input="What is my name and role?",
)
print(second.output_text)   # "You are Priya, and you manage the support team."

stream = client.responses.create(
    model=deployment,
    input="Explain Global Standard vs Data Zone deployments in 4 bullet points.",
    stream=True,
)
for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        usage = event.response.usage
        print(f"\n[{usage.input_tokens} in / {usage.output_tokens} out]")

from pydantic import BaseModel


class Ticket(BaseModel):
    category: str        # "billing" | "shipping" | "technical" | "other"
    urgency: int         # 1 (low) .. 3 (high)
    summary: str


parsed = client.responses.parse(
    model=deployment,
    instructions="Classify the customer message into a support ticket.",
    input="My card was charged twice for order A1002 and I need it fixed today!",
    text_format=Ticket,
)
ticket: Ticket = parsed.output_parsed
print(ticket.model_dump())

embed_client = OpenAI(
    base_url=f"https://{resource_name}.openai.azure.com/openai/v1",
    api_key=token_provider,
)
result = embed_client.embeddings.create(
    model="text-embedding-3-large",      # your embedding deployment name
    input=["refund policy for damaged items", "how do I return a product?"],
)
print(len(result.data), "vectors of", len(result.data[0].embedding), "dimensions")

import random


class RateLimited(Exception):
    def __init__(self, retry_after: float | None):
        self.retry_after = retry_after


def call_with_retry(fn, max_attempts=5, base=1.0, cap=30.0, sleep=print):
    for attempt in range(1, max_attempts + 1):
        try:
            return fn()
        except RateLimited as err:
            if attempt == max_attempts:
                raise
            if err.retry_after is not None:
                wait = err.retry_after                     # the service told us exactly
            else:
                wait = min(cap, base * 2 ** (attempt - 1)) * random.uniform(0.5, 1.0)
            sleep(f"attempt {attempt}: 429, waiting {wait:.2f}s")


random.seed(7)
responses = iter([RateLimited(None), RateLimited(None), RateLimited(4.0), "ok: answer text"])


def fake_model_call():
    item = next(responses)
    if isinstance(item, Exception):
        raise item
    return item


print(call_with_retry(fake_model_call))

# in real code, configure the SDK instead of writing your own loop
client = project.get_openai_client(max_retries=5, timeout=60.0)