AI-103 CERTIFICATION BOOTCAMP

import os
import sys
import time
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import TestingCriterionAzureAIEvaluator
from openai.types.eval_create_params import DataSourceConfigCustom
from openai.types.evals.create_eval_jsonl_run_data_source_param import (
    CreateEvalJSONLRunDataSourceParam, SourceFileContent, SourceFileContentContent)

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

# 1. Fresh answers from the deployment this run just created
queries = ["<a routine question from your work>", "<a hard question from your work>"]
rows = [SourceFileContentContent(item={"query": q, "response": client.responses.create(
    model=model, input=q).output_text}) for q in queries]

# 2. One quality and one safety evaluator, mapped to those two fields
criteria = [TestingCriterionAzureAIEvaluator(
    type="azure_ai_evaluator", name=name, evaluator_name=f"builtin.{name}",
    initialization_parameters={"model": model},
    data_mapping={"query": "{{item.query}}", "response": "{{item.response}}"})
    for name in ("coherence", "violence")]
schema = {"type": "object", "required": ["query", "response"],
          "properties": {"query": {"type": "string"}, "response": {"type": "string"}}}
evaluation = client.evals.create(name="ci-gate", testing_criteria=criteria,
    data_source_config=DataSourceConfigCustom(type="custom", item_schema=schema))
run = client.evals.runs.create(eval_id=evaluation.id, name="ci-run",
    data_source=CreateEvalJSONLRunDataSourceParam(type="jsonl",
        source=SourceFileContent(type="file_content", content=rows)))
while run.status not in ("completed", "failed"):
    time.sleep(5)
    run = client.evals.runs.retrieve(run_id=run.id, eval_id=evaluation.id)
# 3. The gate: an incomplete run, any failure or any error stops the pipeline
print(f"Status: {run.status} | counts: {run.result_counts} | report: {run.report_url}")
if run.status != "completed" or run.result_counts.failed or run.result_counts.errored:
    sys.exit("Evaluation gate FAILED: this change is blocked.")
print("Evaluation gate passed.")

Status: completed | counts: <total, passed, failed and errored counts> | report: <link to the run in the Foundry portal>
Evaluation gate passed.

import os
import time
from pprint import pprint
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import TestingCriterionAzureAIEvaluator
from openai.types.eval_create_params import DataSourceConfigCustom
from openai.types.evals.create_eval_jsonl_run_data_source_param import (
    CreateEvalJSONLRunDataSourceParam, SourceFileContent, SourceFileContentContent)

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

# Prompts that PASSED your guardrail: a blocked prompt has no response to score
prompts = ["<a normal question from your work>", "<a borderline prompt that passed>",
           "<a second borderline prompt that passed>"]
rows = [SourceFileContentContent(item={"query": p, "response": client.responses.create(
    model=model, input=p).output_text}) for p in prompts]

criteria = [TestingCriterionAzureAIEvaluator(
    type="azure_ai_evaluator", name=name, evaluator_name=f"builtin.{name}",
    initialization_parameters={"model": model},
    data_mapping={"query": "{{item.query}}", "response": "{{item.response}}"})
    for name in ("violence", "self_harm", "hate_unfairness")]
schema = {"type": "object", "required": ["query", "response"],
          "properties": {"query": {"type": "string"}, "response": {"type": "string"}}}
evaluation = client.evals.create(name="s23-safety", testing_criteria=criteria,
    data_source_config=DataSourceConfigCustom(type="custom", item_schema=schema))
run = client.evals.runs.create(eval_id=evaluation.id, name="s23-safety-run",
    data_source=CreateEvalJSONLRunDataSourceParam(type="jsonl",
        source=SourceFileContent(type="file_content", content=rows)))
while run.status not in ("completed", "failed"):
    time.sleep(5)
    run = client.evals.runs.retrieve(run_id=run.id, eval_id=evaluation.id)

# Each result carries label, score (0-7), threshold and reason: read every reason
pprint(list(client.evals.runs.output_items.list(run_id=run.id, eval_id=evaluation.id)))
print(f"Status: {run.status} | report: {run.report_url}")

[<one output item per prompt, each with results for violence, self_harm and hate_unfairness: label, score, threshold, reason>]
Status: completed | report: <link to the run in the Foundry portal>

# Sign in and add the Foundry commands to the Azure Developer CLI
azd auth login
azd ext install microsoft.foundry
azd ai project set "$FOUNDRY_PROJECT_ENDPOINT"

# A public, unauthenticated MCP server: the connection stores no secret
azd ai connection create learn-mcp-conn \
    --kind remote-tool \
    --target https://learn.microsoft.com/api/mcp \
    --auth-type none

export MCP_CONNECTION_NAME="learn-mcp-conn"   # as in Learn's MCP sample

import json
import os
from datetime import datetime, timezone
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import PromptAgentDefinition, MCPTool
from openai.types.responses.response_input_param import McpApprovalResponse

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

# Tool-access control: one server, one allowed tool, a person approves every call
tool = MCPTool(server_label="learn-docs", server_url="https://learn.microsoft.com/api/mcp",
               require_approval="always", allowed_tools=["microsoft_docs_search"],
               project_connection_id=os.environ["MCP_CONNECTION_NAME"])
# Provenance: an immutable version, plus who owns it and which change produced it
agent = project.agents.create_version(
    agent_name="audited-agent",
    definition=PromptAgentDefinition(model=os.environ["FOUNDRY_MODEL_NAME"], tools=[tool],
        instructions="Answer Azure questions only from Microsoft Learn, and name the page."),
    description="Session 24: approval-gated Learn search",
    metadata={"owner": "<your-name>", "change_ref": "S24-lab", "risk": "low"},
)
ref = {"agent_reference": {"name": agent.name, "type": "agent_reference"}}
response = openai.responses.create(input="What is a user-assigned managed identity?", extra_body=ref)

decisions = []
for item in response.output:
    if item.type == "mcp_approval_request" and item.id:
        tool_name, args = getattr(item, "name", None), getattr(item, "arguments", None)
        ok = input(f"Approve {tool_name} {args}? (y/N): ").strip().lower() == "y"
        with open("audit.jsonl", "a", encoding="utf-8") as log:   # the approval record
            log.write(json.dumps({"utc": datetime.now(timezone.utc).isoformat(), "approver": "<your-name>",
                "agent": agent.name, "version": agent.version, "response_id": response.id,
                "approval_request_id": item.id, "tool": tool_name, "arguments": args, "approved": ok}) + "\n")
        decisions.append(McpApprovalResponse(type="mcp_approval_response", approve=ok, approval_request_id=item.id))
if decisions:
    response = openai.responses.create(input=decisions, previous_response_id=response.id, extra_body=ref)
print(response.output_text)
project.agents.delete_version(agent_name=agent.name, agent_version=agent.version)

Approve microsoft_docs_search <the search arguments the model proposed>? (y/N): y
<an answer about user-assigned managed identities that names a Learn page>

(second run, answering N)
<the agent's final answer after your refusal>

// In the INDEXER definition: crack images out of files so a skill can read them
"parameters": {
    "configuration": {
        "dataToExtract": "contentAndMetadata",
        "imageAction": "generateNormalizedImages"
    }
}

// In the SKILLSET definition: the OCR skill, one run per normalized image
{
    "@odata.type": "#Microsoft.Skills.Vision.OcrSkill",
    "context": "/document/normalized_images/*",
    "defaultLanguageCode": null,
    "inputs": [
        { "name": "image", "source": "/document/normalized_images/*" }
    ],
    "outputs": [
        { "name": "text", "targetName": "myText" },
        { "name": "layoutText", "targetName": "myLayoutText" }
    ]
}

RG="rg-ai103"
SEARCH_NAME="<your-search-service-name>"
STORAGE_NAME="<your-storage-account-name>"
ACCOUNT_NAME="<your-foundry-resource-name>"
BASE="/subscriptions/<subscription-id>/resourceGroups/$RG/providers"
FOUNDRY_SCOPE="$BASE/Microsoft.CognitiveServices/accounts/$ACCOUNT_NAME"

# Basic: the lowest tier with managed identity support. Billed hourly until deleted.
az search service create --name $SEARCH_NAME --resource-group $RG \
    --sku basic --partition-count 1 --replica-count 1 \
    --identity-type SystemAssigned

# Object IDs: the search service's identity, and you
SEARCH_MI=$(az ad sp list --display-name "$SEARCH_NAME" --query [].id --output tsv)
ME=$(az ad signed-in-user show --query id --output tsv)

# Search identity: read blobs, call the embedding model, bill skills without a key
az role assignment create --assignee-object-id $SEARCH_MI --assignee-principal-type ServicePrincipal \
    --role "Storage Blob Data Reader" --scope "$BASE/Microsoft.Storage/storageAccounts/$STORAGE_NAME"
az role assignment create --assignee-object-id $SEARCH_MI --assignee-principal-type ServicePrincipal \
    --role "Cognitive Services OpenAI User" --scope $FOUNDRY_SCOPE
az role assignment create --assignee-object-id $SEARCH_MI --assignee-principal-type ServicePrincipal \
    --role "Cognitive Services User" --scope $FOUNDRY_SCOPE

# You: manage search objects and query the index; call Content Understanding
for ROLE in "Search Service Contributor" "Search Index Data Contributor" "Search Index Data Reader"; do
    az role assignment create --assignee-object-id $ME --role "$ROLE" \
        --scope "$BASE/Microsoft.Search/searchServices/$SEARCH_NAME"
done
az role assignment create --assignee-object-id $ME --role "Cognitive Services User" --scope $FOUNDRY_SCOPE

import os
import time
from azure.identity import DefaultAzureCredential
from azure.search.documents.indexes import SearchIndexerClient

client = SearchIndexerClient(
    endpoint=os.environ["AZURE_SEARCH_SERVICE_ENDPOINT"],  # https://<service-name>.search.windows.net
    credential=DefaultAzureCredential(),
)
name = os.environ["AZURE_SEARCH_INDEXER_NAME"]            # my variable: the indexer the wizard made

# What was built: data source, target index, and every skill in the skillset
indexer = client.get_indexer(name)
print(f"Data source: {indexer.data_source_name} -> index: {indexer.target_index_name}")
for skill in client.get_skillset(indexer.skillset_name).skills:
    print(f"  skill: {skill.odata_type}  context: {skill.context}")

# Run again to pick up new files (only once the previous run has finished)
client.run_indexer(name)
time.sleep(90)

# How the latest run went, item by item
status = client.get_indexer_status(name)
last = status.last_result
print(f"Indexer: {status.status}")
if last:
    print(f"Last run: {last.status}  items: {last.item_count}  failed: {last.failed_item_count}")
    for error in last.errors:
        print(f"  ERROR   {error.key}: {error.error_message}")
    for warning in last.warnings:
        print(f"  WARNING {warning.key}: {warning.message}")

Data source: <data source name> -> index: <index name>
  skill: <text split skill type>  context: <...>
  skill: <embedding skill type>  context: <...>
  skill: #Microsoft.Skills.Vision.OcrSkill  context: <...>
  <any other skills the wizard added>
Indexer: running
Last run: success  items: <your file count>  failed: 0
(if the last run says inProgress, wait and run the script again)