Microsoft Foundry Series (Bonus) — Hosted Agents and Multi-Agent Workflows with Microsoft Agent Framework

1python -m venv .venv
2source .venv/bin/activate
3pip install agent-framework-core agent-framework-foundry agent-framework-orchestrations \
4  azure-identity python-dotenv "psycopg[binary]"
5pip install --pre agent-framework-foundry-hosting

1 "SKU ..." ─► Researcher ─► Writer ─► capture_draft ─► Compliance ─► review_gate
2  (catalog + web search)     ▲   (ProductDescription)   (brand rules)     │
3                             └──────────── rejected (max 2×) ─────────────┤
4                                                                approved  ▼
5                                    publish ◄── de-DE / fr-FR / nl-NL ◄── localize
6                                  (fan-in, JSON)   (in parallel)          (fan-out)

 1# content_workflow/tools.py
 2import json
 3import os
 4from typing import Annotated
 5
 6import psycopg
 7from agent_framework import tool
 8from azure.identity import DefaultAzureCredential
 9from pydantic import BaseModel, Field
10
11from app.models import ProductDescription  # unchanged Part 5 model
12
13credential = DefaultAzureCredential()
14
15
16class ReviewResult(BaseModel):
17    approved: bool
18    issues: list[str] = Field(description="Concrete, fixable problems; empty if approved")
19
20
21@tool(approval_mode="never_require")
22def get_product_details(
23    sku: Annotated[str, Field(description="Product SKU, e.g. KB-MEC-2026")],
24) -> str:
25    """Fetch product facts (name, price, features, stock) from the catalog database."""
26    token = credential.get_token("https://ossrdbms-aad.database.windows.net/.default").token
27    with psycopg.connect(
28        host=os.environ["PGHOST"],
29        dbname=os.environ.get("PGDATABASE", "catalog"),
30        user=os.environ["PGUSER"],  # Entra principal name mapped in PostgreSQL
31        password=token,
32        sslmode="require",
33    ) as conn:
34        row = conn.execute(
35            "SELECT name, category, price::float, currency, stock_quantity, specs, features "
36            "FROM products WHERE sku = %s",
37            (sku.strip().upper(),),
38        ).fetchone()
39    if row is None:
40        return json.dumps({"error": f"SKU {sku} not found"})
41    keys = ("name", "category", "price", "currency", "stock_quantity", "specs", "features")
42    return json.dumps({"sku": sku, **dict(zip(keys, row))})

 1# content_workflow/agents.py
 2import os
 3
 4from agent_framework import Agent
 5from agent_framework.foundry import FoundryChatClient
 6from azure.identity import DefaultAzureCredential
 7
 8from app.models import ProductDescription
 9
10from .tools import ReviewResult, get_product_details
11
12BRAND_RULES = """- Tone: confident, friendly, no hype words ("revolutionary", "best ever").
13- Never claim health, safety or environmental benefits that are not in the catalog facts.
14- Never mention competitors by name.
15- Price and specs must match the catalog facts exactly.
16- No content that is violent, hateful, sexual or encourages self-harm (Part 9 policy)."""
17
18
19def _client(model_env: str) -> FoundryChatClient:
20    return FoundryChatClient(
21        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
22        model=os.environ[model_env],
23        credential=DefaultAzureCredential(),
24    )
25
26
27def create_agents() -> dict[str, Agent]:
28    small = _client("SMALL_MODEL")    # gpt-4.1-mini
29    large = _client("WRITER_MODEL")   # gpt-4.1
30
31    researcher = Agent(
32        client=small,
33        name="researcher",
34        instructions=(
35            "You research products for copywriters. Always call get_product_details first. "
36            "Then use web search for 2-3 current buying trends or keywords for the category. "
37            "Return concise bullet notes: FACTS (from catalog only) and TRENDS (with sources). "
38            "Never invent specifications."
39        ),
40        tools=[get_product_details, FoundryChatClient.get_web_search_tool(search_context_size="low")],
41        default_options={"store": False},
42    )
43
44    writer = Agent(
45        client=large,
46        name="writer",
47        instructions=(
48            "You are an e-commerce copywriter. Write an SEO-friendly product description "
49            "with locale en-GB using ONLY the facts provided. Weave in trends where relevant.\n"
50            f"Brand rules:\n{BRAND_RULES}"
51        ),
52        default_options={"response_format": ProductDescription, "store": False},
53    )
54
55    compliance = Agent(
56        client=small,
57        name="compliance",
58        instructions=(
59            "You are a strict brand and compliance reviewer. Check the product description "
60            f"against these rules:\n{BRAND_RULES}\n"
61            "Approve only if every rule passes. Otherwise list each issue as a concrete fix."
62        ),
63        default_options={"response_format": ReviewResult, "store": False},
64    )
65
66    translators = {
67        f"translator_{locale[:2]}": Agent(
68            client=small,
69            name=f"translator_{locale[:2]}",
70            instructions=(
71                f"Translate the product description JSON into {locale} and set locale to {locale}. Keep the same structure, "
72                "localise idioms and units, keep brand and product names untranslated."
73            ),
74            default_options={"response_format": ProductDescription, "store": False},
75        )
76        for locale in ("de-DE", "fr-FR", "nl-NL")
77    }
78
79    return {"researcher": researcher, "writer": writer, "compliance": compliance, **translators}

 1# content_workflow/workflow.py
 2import json
 3from typing import Never
 4
 5from agent_framework import (
 6    AgentExecutorRequest,
 7    AgentExecutorResponse,
 8    Message,
 9    Workflow,
10    WorkflowBuilder,
11    WorkflowContext,
12    executor,
13)
14
15from app.models import ProductDescription
16
17from .agents import create_agents
18from .tools import ReviewResult
19
20MAX_REVISIONS = 2
21
22
23@executor(id="capture_draft")
24async def capture_draft(response: AgentExecutorResponse, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
25    draft = ProductDescription.model_validate_json(response.agent_response.text)
26    ctx.set_state("draft", draft.model_dump_json())
27    # The reviewer only sees the draft, not the whole research conversation: fewer tokens.
28    await ctx.send_message(AgentExecutorRequest(
29        messages=[Message("user", [f"Review this product description:\n{draft.model_dump_json()}"])],
30        should_respond=True,
31    ))
32
33
34@executor(id="review_gate")
35async def review_gate(response: AgentExecutorResponse, ctx: WorkflowContext[AgentExecutorRequest | str]) -> None:
36    review = ReviewResult.model_validate_json(response.agent_response.text)
37    revisions = ctx.get_state("revisions", 0)
38    draft = ctx.get_state("draft")
39
40    if review.approved or revisions >= MAX_REVISIONS:
41        ctx.set_state("review", review.model_dump())
42        await ctx.send_message(draft, target_id="localize")
43        return
44
45    ctx.set_state("revisions", revisions + 1)
46    feedback = "\n".join(f"- {issue}" for issue in review.issues)
47    await ctx.send_message(
48        AgentExecutorRequest(
49            messages=[Message("user", [f"Revise this draft. Fix these issues:\n{feedback}\n\nDraft:\n{draft}"])],
50            should_respond=True,
51        ),
52        target_id="writer",
53    )
54
55
56@executor(id="localize")
57async def localize(draft_json: str, ctx: WorkflowContext[AgentExecutorRequest]) -> None:
58    await ctx.send_message(AgentExecutorRequest(messages=[Message("user", [draft_json])], should_respond=True))
59
60
61@executor(id="publish")
62async def publish(results: list[AgentExecutorResponse], ctx: WorkflowContext[Never, str]) -> None:
63    draft = ProductDescription.model_validate_json(ctx.get_state("draft"))
64    translations = [ProductDescription.model_validate_json(r.agent_response.text) for r in results]
65    payload = {
66        "locales": {d.locale: d.model_dump() for d in [draft, *translations]},  # en-GB, de-DE, ...
67        "review": ctx.get_state("review"),
68        "revisions": ctx.get_state("revisions", 0),
69    }
70    await ctx.yield_output(json.dumps(payload, ensure_ascii=False))
71
72
73def build_workflow() -> Workflow:
74    a = create_agents()
75    translators = [a["translator_de"], a["translator_fr"], a["translator_nl"]]
76    return (
77        WorkflowBuilder(name="product-content-v1", start_executor=a["researcher"], output_from=[publish])
78        .add_edge(a["researcher"], a["writer"])
79        .add_edge(a["writer"], capture_draft)
80        .add_edge(capture_draft, a["compliance"])
81        .add_edge(a["compliance"], review_gate)
82        .add_edge(review_gate, a["writer"])        # revision loop
83        .add_edge(review_gate, localize)           # approved path
84        .add_fan_out_edges(localize, translators)
85        .add_fan_in_edges(translators, publish)
86        .build()
87    )

 1# run_local.py
 2import asyncio
 3import json
 4
 5from dotenv import load_dotenv
 6
 7from content_workflow.workflow import build_workflow
 8
 9load_dotenv()
10
11
12async def main() -> None:
13    workflow = build_workflow()
14    async for event in workflow.run("Create product content for SKU KB-MEC-2026", stream=True):
15        if event.type == "executor_invoked":
16            print(f"▶ {event.executor_id}")
17        elif event.type == "output":
18            print(json.dumps(json.loads(event.data), indent=2, ensure_ascii=False))
19
20
21if __name__ == "__main__":
22    asyncio.run(main())

1▶ researcher  ▶ writer  ▶ capture_draft  ▶ compliance  ▶ review_gate
2▶ writer      ← reviewer flagged "best keyboard ever" as hype
3▶ capture_draft  ▶ compliance  ▶ review_gate  ▶ localize
4▶ translator_de  ▶ translator_fr  ▶ translator_nl  ▶ publish
5{ "locales": { "en-GB": { "sku": "KB-MEC-2026", "title": "ProType Mechanical Keyboard", ... }, "de-DE": { ... }, ... }, "revisions": 1, ... }

 1# main.py
 2from agent_framework import Workflow
 3from agent_framework_foundry_hosting import HostedResponseRequest, ResponsesHostServer, WorkflowTurn
 4from dotenv import load_dotenv
 5
 6from content_workflow.workflow import build_workflow
 7
 8
 9def workflow_factory(request: HostedResponseRequest) -> Workflow:
10    return build_workflow()
11
12
13async def parse_response(request: HostedResponseRequest) -> WorkflowTurn[str]:
14    return WorkflowTurn(input=await request.get_input_text() or "")
15
16
17def main() -> None:
18    load_dotenv()
19    ResponsesHostServer(workflow=workflow_factory, parse_response=parse_response).run()
20
21
22if __name__ == "__main__":
23    main()

 1FROM python:3.13-slim
 2
 3ENV PYTHONNOUSERSITE=1 \
 4    PYTHONUNBUFFERED=1
 5
 6WORKDIR /app
 7COPY requirements.txt .
 8RUN pip install --no-cache-dir -r requirements.txt
 9COPY . .
10
11EXPOSE 8088
12CMD ["python", "main.py"]

 1# yaml-language-server: $schema=https://raw.githubusercontent.com/Azure/azure-dev/main/schemas/v1.0/azure.yaml.json
 2name: product-content
 3requiredVersions:
 4  azd: ">=1.36.0"
 5  extensions:
 6    azure.ai.agents: ">=1.0.0-beta.20"
 7services:
 8  ai-project:
 9    host: azure.ai.project
10    endpoint: ${AZURE_AI_PROJECT_ENDPOINT}      # existing project from Parts 1-2
11  product-content-workflow:
12    host: azure.ai.agent
13    project: .
14    language: docker
15    uses:
16      - ai-project
17    kind: hosted
18    name: product-content-workflow
19    description: Researcher -> Writer -> Compliance loop -> parallel translators
20    startupCommand: python main.py
21    protocols:
22      - protocol: responses
23        version: 2.0.0
24    env:
25      WRITER_MODEL: gpt-4.1
26      SMALL_MODEL: gpt-4.1-mini
27      PGHOST: ${PGHOST}
28      PGUSER: product-content-workflow
29      PYTHONNOUSERSITE: "1"
30    container:
31      resources:
32        cpu: "1"
33        memory: 2Gi

 1# One-time: the Foundry extension for azd
 2azd ext install azure.ai.agents
 3azd auth login
 4
 5# Local inner loop: runs main.py on http://localhost:8088 and opens the Agent Inspector
 6azd ai agent run
 7azd ai agent invoke --local "Create product content for SKU KB-MEC-2026"
 8
 9# Build in ACR, create a new immutable agent version, wire identity and RBAC
10azd deploy
11
12azd ai agent show                      # name, version, protocols, endpoints
13azd ai agent invoke "Create product content for SKU TB-PRO-2026"
14azd ai agent monitor --follow          # live container logs

 1@description('Role definition ID (GUID) for "Container Registry Repository Reader"')
 2param acrRepositoryReaderRoleId string
 3
 4@description('Role definition ID (GUID) for "Foundry Agent Consumer"')
 5param agentConsumerRoleId string
 6
 7@description('Principal ID of foundry-demo-api-id, the Container App identity (Part 6)')
 8param apiPrincipalId string
 9
10resource acr 'Microsoft.ContainerRegistry/registries@2023-07-01' existing = {
11  name: replace('${baseName}acr${uniqueString(resourceGroup().id)}', '-', '')
12}
13
14resource projectAcrPull 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
15  scope: acr
16  name: guid(acr.id, project.id, acrRepositoryReaderRoleId)
17  properties: {
18    principalId: project.identity.principalId
19    principalType: 'ServicePrincipal'
20    roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', acrRepositoryReaderRoleId)
21  }
22}
23
24resource apiCanInvokeAgents 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
25  scope: project
26  name: guid(project.id, apiPrincipalId, agentConsumerRoleId)
27  properties: {
28    principalId: apiPrincipalId
29    principalType: 'ServicePrincipal'
30    roleDefinitionId: subscriptionResourceId('Microsoft.Authorization/roleDefinitions', agentConsumerRoleId)
31  }
32}

1az role definition list --name "Foundry Agent Consumer" --query "[0].name" -o tsv
2az role definition list --name "Container Registry Repository Reader" --query "[0].name" -o tsv

 1AGENT_OID=$(az rest --method GET \
 2  --url "${PROJECT_ENDPOINT}/agents/product-content-workflow?api-version=v1" \
 3  --resource "https://ai.azure.com" \
 4  --query "instance_identity.principal_id" -o tsv)
 5
 6# As the PostgreSQL Entra admin: create the principal (in the postgres DB), then grant (in catalog)
 7psql "host=$PGHOST dbname=postgres user=$PG_ADMIN sslmode=require" -c \
 8  "SELECT * FROM pgaadauth_create_principal_with_oid('product-content-workflow', '$AGENT_OID', 'service', false, false);"
 9psql "host=$PGHOST dbname=catalog user=$PG_ADMIN sslmode=require" -c \
10  'GRANT CONNECT ON DATABASE catalog TO "product-content-workflow"; GRANT SELECT ON products TO "product-content-workflow";'

 1# app/foundry.py (additions)
 2    # inside lifespan(), next to app.state.openai
 3    app.state.content_agent = project.get_openai_client(agent_name="product-content-workflow")
 4
 5
 6def get_content_agent(request: Request) -> AsyncOpenAI:
 7    return request.app.state.content_agent
 8
 9
10# app/main.py (addition)
11ContentAgentDep = Annotated[AsyncOpenAI, Depends(get_content_agent)]
12
13
14@app.post("/products/{sku}/content", response_model=ProductDescription)
15async def generate_content(
16    sku: str, agent: ContentAgentDep, limiter: LimiterDep, locale: Locale = "en-GB"
17) -> ProductDescription:
18    async with limiter:
19        response = await agent.responses.create(input=f"Create product content for SKU {sku}")
20    try:
21        content = json.loads(response.output_text)
22        return ProductDescription.model_validate(content["locales"][locale])
23    except (json.JSONDecodeError, KeyError, ValueError):
24        raise HTTPException(status_code=502, detail="Workflow returned unexpected output")

 1# eval_workflow.py
 2import asyncio
 3import os
 4
 5from agent_framework import evaluate_workflow
 6from agent_framework.foundry import FoundryChatClient, FoundryEvals
 7from azure.identity import DefaultAzureCredential
 8
 9from content_workflow.workflow import build_workflow
10
11
12async def main() -> None:
13    judge = FoundryChatClient(
14        project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
15        model=os.environ["WRITER_MODEL"],
16        credential=DefaultAzureCredential(),
17    )
18    results = await evaluate_workflow(
19        workflow=build_workflow(),
20        queries=[
21            "Create product content for SKU KB-MEC-2026",
22            "Create product content for SKU TB-PRO-2026",
23        ],
24        evaluators=FoundryEvals(
25            client=judge,
26            evaluators=[
27                FoundryEvals.TASK_ADHERENCE,
28                FoundryEvals.TOOL_CALL_ACCURACY,
29                FoundryEvals.INTENT_RESOLUTION,
30            ],
31        ),
32    )
33    for r in results:
34        print(f"Overall: {r.passed}/{r.total}  {r.report_url}")
35        for agent_name, agent_eval in r.sub_results.items():
36            print(f"  {agent_name}: {agent_eval.passed}/{agent_eval.total}")
37
38
39if __name__ == "__main__":
40    asyncio.run(main())

1az rest --method DELETE \
2  --url "${PROJECT_ENDPOINT}/agents/product-content-workflow?api-version=v1&force=true" \
3  --resource "https://ai.azure.com"

1az group delete --name rg-foundry-demo --yes --no-wait