AI-103 CERTIFICATION BOOTCAMP

import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import PromptAgentDefinition, FunctionTool

INSTRUCTIONS = """Role: You are the parcel-tracking assistant for a Lagos logistics firm.
Goals: 1) Tell customers where their parcel is. 2) Set honest expectations.
3) Hand over to a human colleague when you are unsure.
Tools: Call get_parcel_status whenever the customer gives a tracking ID.
If the ID is missing, ask for it. Never guess a status.
Limits: Never promise refunds or delivery dates that the tool did not return.
Style: Plain English, three sentences or fewer."""

status_tool = FunctionTool(
    name="get_parcel_status",
    description="Look up the latest status of a parcel by its tracking ID.",
    parameters={
        "type": "object",
        "properties": {
            "tracking_id": {"type": "string", "description": "Tracking ID, for example LOS-48213"},
            "detail": {"type": "string", "enum": ["summary", "full"], "description": "How much history to return"},
        },
        "required": ["tracking_id", "detail"],
        "additionalProperties": False,
    },
    strict=True,
)

project = AIProjectClient(endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], credential=DefaultAzureCredential())
agent = project.agents.create_version(
    agent_name="parcel-agent",
    definition=PromptAgentDefinition(
        model=os.environ["FOUNDRY_MODEL_NAME"],  # your deployment name
        instructions=INSTRUCTIONS,
        tools=[status_tool],
    ),
)
print(f"Agent created (name: {agent.name}, version: {agent.version})")

Agent created (name: parcel-agent, version: 1)

import json
import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from openai.types.responses.response_input_param import FunctionCallOutput

def get_parcel_status(tracking_id: str, detail: str) -> str:
    # Illustrative stand-in for the firm's tracking system
    return json.dumps({"tracking_id": tracking_id, "status": "At Abuja hub", "updated": "09:40"})

project = AIProjectClient(endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], credential=DefaultAzureCredential())
openai = project.get_openai_client()
agent = {"agent_reference": {"name": "parcel-agent", "type": "agent_reference"}}

conversation = openai.conversations.create()  # durable, server-side history
print("Conversation:", conversation.id)

for text in ["Where is my parcel?", "Sorry, the tracking ID is LOS-48213."]:
    print("User:", text)
    response = openai.responses.create(conversation=conversation.id, input=text, extra_body=agent)
    outputs = []
    for item in response.output:
        print(f"  [{item.type}]")
        if item.type == "function_call" and item.name == "get_parcel_status":
            print(f"  -> {item.name}({item.arguments})")
            outputs.append(FunctionCallOutput(
                type="function_call_output",
                call_id=item.call_id,
                output=get_parcel_status(**json.loads(item.arguments)),
            ))
    if outputs:  # your app ran the function: hand the result back in the same conversation
        response = openai.responses.create(conversation=conversation.id, input=outputs, extra_body=agent)
    print("Agent:", response.output_text)

Conversation: <conversation-id>
User: Where is my parcel?
  [message]
Agent: <asks you for the tracking ID>
User: Sorry, the tracking ID is LOS-48213.
  [function_call]
  -> get_parcel_status({"tracking_id":"LOS-48213","detail":"summary"})
Agent: <says the parcel is at the Abuja hub, updated 09:40>

import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import PromptAgentDefinition, FileSearchTool, FunctionTool

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

# Knowledge: a vector store holding your own (illustrative) policy file
vector_store = openai.vector_stores.create(name="leave-policy-store")
with open("leave_policy.md", "rb") as file_handle:
    vs_file = openai.vector_stores.files.upload_and_poll(vector_store_id=vector_store.id, file=file_handle)

# Action: a function your app runs
balance_tool = FunctionTool(
    name="get_leave_balance",
    description="Return the remaining annual leave days for a staff ID such as E1042.",
    parameters={
        "type": "object",
        "properties": {"staff_id": {"type": "string", "description": "Staff ID, for example E1042"}},
        "required": ["staff_id"],
        "additionalProperties": False,
    },
    strict=True,
)

agent = project.agents.create_version(
    agent_name="hr-leave-agent",
    definition=PromptAgentDefinition(
        model=os.environ["FOUNDRY_MODEL_NAME"],
        instructions="You answer staff questions about annual leave. Use File Search for any policy "
                     "rule and say which rule you used. Use get_leave_balance only when the user gives "
                     "a staff ID. If the policy does not cover the question, say you don't know.",
        tools=[FileSearchTool(vector_store_ids=[vector_store.id]), balance_tool],
    ),
)
print(f"Agent {agent.name} v{agent.version} | vector store {vector_store.id} | file {vs_file.id}")

Agent hr-leave-agent v1 | vector store <vector-store-id> | file <file-id>

import os
from azure.identity import DefaultAzureCredential
from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import (
    MemorySearchPreviewTool,
    MemoryStoreDefaultDefinition,
    MemoryStoreDefaultOptions,
    PromptAgentDefinition,
)

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

store = project.beta.memory_stores.create(
    name="hr_memory_store",
    definition=MemoryStoreDefaultDefinition(
        chat_model=chat_model,
        embedding_model=os.environ["MEMORY_STORE_EMBEDDING_MODEL_DEPLOYMENT_NAME"],
        options=MemoryStoreDefaultOptions(
            chat_summary_enabled=True,
            user_profile_enabled=True,
            user_profile_details="Avoid sensitive data, such as age, financials, precise location, and credentials",
        ),
    ),
    description="Long-term memory for the HR leave agent",
)

agent = project.agents.create_version(
    agent_name="hr-memory-agent",
    definition=PromptAgentDefinition(
        model=chat_model,
        instructions="You help staff with leave questions and remember the preferences they state.",
        # "{{$userId}}" isolates memories per end user; update_delay=1 is for the demo only (default 300)
        tools=[MemorySearchPreviewTool(memory_store_name=store.name, scope="{{$userId}}", update_delay=1)],
    ),
)
print(f"Memory store {store.name} | agent {agent.name} v{agent.version}")

Memory store hr_memory_store | agent hr-memory-agent v1

# In your course virtual environment (Python 3.10+)
pip install "azure-ai-projects>=2.4.0" azure-identity jsonref   # 2.4.0: the floor since Session 9

# Learn's first-success flow: save the sample spec into an assets folder
mkdir -p assets
# Download weather_openapi.json from the link on Learn's OpenAPI how-to into ./assets/

az login
export FOUNDRY_PROJECT_ENDPOINT="https://<resource-name>.services.ai.azure.com/api/projects/<project-name>"
export FOUNDRY_MODEL_NAME="<your-deployment-name>"

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

ALLOWED = {"BILLING", "CARD", "TRANSFER", "OTHER"}  # rule: the only routes the bank accepts
ACCOUNT_NO = re.compile(r"\b\d{10}\b")             # rule: 10-digit account numbers never reach the model
PROMPT = ('Classify this complaint. Reply with JSON only, for example '
          '{"route": "BILLING", "reason": "short reason"}. Routes: BILLING, CARD, TRANSFER, OTHER.')

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

def route(ticket: str) -> dict:
    trace = []
    if "fraud" in ticket.lower():                                     # rule before: hard stop
        trace.append("pre-rule: fraud keyword -> FRAUD_TEAM, model not called")
        return {"route": "FRAUD_TEAM", "trace": trace}
    clean, count = ACCOUNT_NO.subn("[ACCOUNT]", ticket)               # rule before: redact
    trace.append(f"pre-rule: redacted {count} account number(s)")
    r = openai.responses.create(model=os.environ["FOUNDRY_MODEL_NAME"], input=f"{PROMPT}\n\n{clean}")
    trace.append(f"model: {r.output_text}")
    try:                                                              # rule after: validate
        result = json.loads(r.output_text)
    except json.JSONDecodeError:
        result = {}
    if not isinstance(result, dict) or result.get("route") not in ALLOWED:
        trace.append("post-rule: output not allowed -> HUMAN_QUEUE")
        return {"route": "HUMAN_QUEUE", "trace": trace}
    trace.append("post-rule: accepted")
    return {"route": result["route"], "trace": trace}

for ticket in ["I think this is fraud: a card payment I never made.",
               "Transfer from 0123456789 failed but the money left my account."]:
    print(json.dumps(route(ticket), indent=2))

{
  "route": "FRAUD_TEAM",
  "trace": [
    "pre-rule: fraud keyword -> FRAUD_TEAM, model not called"
  ]
}
{
  "route": "TRANSFER",
  "trace": [
    "pre-rule: redacted 1 account number(s)",
    "model: {\"route\": \"TRANSFER\", \"reason\": \"<short reason>\"}",
    "post-rule: accepted"
  ]
}