Foundry SDK deep dive

# One package for the platform, one for identity.
pip install "azure-ai-projects>=2.3.0" azure-identity

# Optional, depending on what you touch:
pip install azure-ai-evaluation          # evaluators
pip install azure-search-documents       # Azure AI Search
pip install azure-cognitiveservices-speech
pip install azure-monitor-opentelemetry  # tracing

# Authenticate locally before running anything.
az login

import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient

PROJECT_ENDPOINT = os.environ["FOUNDRY_PROJECT_ENDPOINT"]
# https://<resource-name>.services.ai.azure.com/api/projects/<project-name>

project = AIProjectClient(
    endpoint=PROJECT_ENDPOINT,
    credential=DefaultAzureCredential(),
)

openai = project.get_openai_client()

# 1. Simplest call
r = openai.responses.create(model="gpt-5-mini", input="Define RAG in one sentence.")
print(r.output_text)

# 2. With roles, when you need a system instruction
r = openai.responses.create(
    model="gpt-5-mini",
    input=[
        {"role": "system", "content": "You answer only about Azure. Be concise."},
        {"role": "user", "content": "What is a Foundry project?"},
    ],
)

# 3. Chained turns without a conversation object
first = openai.responses.create(model="gpt-5-mini", input="Summarise this contract.")
second = openai.responses.create(
    model="gpt-5-mini",
    previous_response_id=first.id,
    input="Now list the three biggest risks.",
)

# 4. Multi-turn with server-side state
conversation = openai.conversations.create()
openai.responses.create(conversation=conversation.id, model="gpt-5-mini", input="Book a room in Lisbon.")
openai.responses.create(conversation=conversation.id, model="gpt-5-mini", input="Make it two nights.")

# 5. Tuning parameters
r = openai.responses.create(
    model="gpt-5-mini",
    input=prompt,
    temperature=0,            # deterministic-ish; do not also tune top_p
    max_output_tokens=800,    # also feeds the TPM rate-limit estimate
)

# Streaming improves perceived latency; the event types matter.
with openai.responses.stream(model="gpt-5-mini", input="Write a haiku about Azure.") as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
        elif event.type == "response.error":
            raise RuntimeError(event.error)

    final = stream.get_final_response()

print()
print(final.usage.input_tokens, final.usage.output_tokens)

from pydantic import BaseModel
from typing import Literal

class Extraction(BaseModel):
    customer: str
    order_id: str
    issue_type: Literal["delivery", "billing", "quality", "other"]
    urgent: bool

parsed = openai.responses.parse(
    model="gpt-5-mini",
    input=[{"role": "user", "content": email_text}],
    text_format=Extraction,       # constrains generation to the schema
)

record = parsed.output_parsed     # a typed Extraction, not a string
print(record.issue_type, record.urgent)

import json

TOOLS = [{
    "type": "function",
    "name": "get_order_status",
    "description": "Look up the delivery status of an order by its id.",
    "parameters": {
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
        "additionalProperties": False,
    },
}]

def get_order_status(order_id: str) -> dict:
    return {"order_id": order_id, "status": "in transit", "eta": "2026-09-19"}

response = openai.responses.create(
    model="gpt-5-mini",
    input="Where is order A-4471?",
    tools=TOOLS,
)

# The model returns tool calls; YOUR code executes them and feeds results back.
# Loop, because one turn can produce several calls or chain into more.
while True:
    calls = [item for item in response.output if item.type == "function_call"]
    if not calls:
        break

    outputs = []
    for call in calls:
        args = json.loads(call.arguments)
        result = get_order_status(**args)
        outputs.append({
            "type": "function_call_output",
            "call_id": call.call_id,        # must match the call
            "output": json.dumps(result),
        })

    response = openai.responses.create(
        model="gpt-5-mini",
        previous_response_id=response.id,
        input=outputs,
        tools=TOOLS,
    )

print(response.output_text)

# File search over a Foundry-managed vector store
file = openai.files.create(file=open("handbook.pdf", "rb"), purpose="assistants")
store = openai.vector_stores.create(name="hr-handbook", file_ids=[file.id])

r = openai.responses.create(
    model="gpt-5-mini",
    input="How many days of parental leave do contractors get?",
    tools=[{"type": "file_search", "vector_store_ids": [store.id]}],
)

# Code interpreter: the model writes and runs code in a sandbox
r = openai.responses.create(
    model="gpt-5-mini",
    input="Chart the monthly totals in this CSV and describe the trend.",
    tools=[{"type": "code_interpreter"}],
)

# A remote MCP server as a tool
r = openai.responses.create(
    model="gpt-5-mini",
    input="Which work items are assigned to me and still open?",
    tools=[{
        "type": "mcp",
        "server_label": "azure-devops",
        "server_url": "https://example.com/mcp",
        "require_approval": "always",     # human gate before tool execution
    }],
)

from azure.ai.projects.models import PromptAgentDefinition

# Create (or update) an agent. Each call produces a new immutable VERSION.
agent = project.agents.create_version(
    agent_name="support-triage",
    definition=PromptAgentDefinition(
        model="gpt-5-mini",
        instructions=(
            "Triage incoming support requests. Classify severity, identify the "
            "product area, and create a ticket only after confirming with the user."
        ),
        tools=[{"type": "file_search", "vector_store_ids": [store.id]}],
    ),
)
print(agent.id, agent.name, agent.version)

# Talk to it: get a client pre-bound to the agent.
agent_openai = project.get_openai_client(agent_name="support-triage")
conversation = agent_openai.conversations.create()

reply = agent_openai.responses.create(
    conversation=conversation.id,
    input="Checkout has been throwing 500s for twenty minutes.",
)
print(reply.output_text)

# Housekeeping
for version in project.agents.list_versions(agent_name="support-triage"):
    print(version.version, version.created_at)

from azure.ai.evaluation import (
    GroundednessEvaluator, RelevanceEvaluator, CoherenceEvaluator, evaluate,
)

judge = {"azure_endpoint": AZURE_ENDPOINT, "azure_deployment": "gpt-5-mini"}

results = evaluate(
    data="golden.jsonl",   # rows of {query, context, response, ground_truth}
    evaluators={
        "groundedness": GroundednessEvaluator(judge),
        "relevance": RelevanceEvaluator(judge),
        "coherence": CoherenceEvaluator(judge),
    },
    azure_ai_project=PROJECT_ENDPOINT,   # publish results to the portal
    output_path="./eval_results.json",
)

print(results["metrics"])
assert results["metrics"]["groundedness.groundedness"] >= 4.0   # CI gate

import os
from azure.monitor.opentelemetry import configure_azure_monitor

connection_string = project.telemetry.get_application_insights_connection_string()
configure_azure_monitor(connection_string=connection_string)

# Prompt and completion text is NOT recorded by default: it can contain
# customer data. Opt in deliberately, and only where retention allows.
os.environ["AZURE_TRACING_GEN_AI_CONTENT_RECORDING_ENABLED"] = "true"

# From here, model calls, tool calls and agent decisions appear as spans in
# Application Insights and in the Foundry portal's tracing view.

from azure.identity import DefaultAzureCredential
from azure.search.documents import SearchClient
from azure.search.documents.models import VectorizableTextQuery

client = SearchClient(
    endpoint="https://my-search.search.windows.net",
    index_name="policies",
    credential=DefaultAzureCredential(),   # keyless; needs a Search data-plane role
)

results = client.search(
    search_text=question,                            # BM25
    vector_queries=[VectorizableTextQuery(           # ANN, embedded by the index vectorizer
        text=question, k_nearest_neighbors=50, fields="content_vector",
    )],
    query_type="semantic",                           # L2 rerank
    semantic_configuration_name="default",
    query_caption="extractive",
    filter="department eq 'HR'",                     # text fields only
    top=5,
)