Python: Update workflow orchestration samples to use AzureOpenAIResponsesClient (#4285)

* Update workflow orchestration samples to use AzureOpenAIResponsesClient

* Fix broken link
This commit is contained in:
Evan Mattson
2026-02-26 16:51:25 +09:00
committed by GitHub
Unverified
parent cfaa0c6283
commit 8e2cc4bedc
18 changed files with 192 additions and 70 deletions
+1 -1
View File
@@ -498,7 +498,7 @@ We need to decide what AIContent types, each agent response type will be mapped
| Google ADK | **Approach 1** Both [input and output schemas can be specified for LLM Agents](https://google.github.io/adk-docs/agents/llm-agents/#structuring-data-input_schema-output_schema-output_key) at construction time. This option is specific to this agent type and other agent types do not necessarily support |
| AWS (Strands) | **Approach 2** Supports a special invocation method called [structured_output](https://strandsagents.com/latest/documentation/docs/api-reference/python/agent/agent/#strands.agent.agent.Agent.structured_output) |
| LangGraph | **Approach 1** Supports [configuring an agent](https://langchain-ai.github.io/langgraph/agents/agents/?h=structured#6-configure-structured-output) at agent construction time, and a [structured response](https://langchain-ai.github.io/langgraph/agents/run_agents/#output-format) can be retrieved as a special property on the agent response |
| Agno | **Approach 1** Supports [configuring an agent](https://docs.agno.com/examples/getting-started/structured-output) at agent construction time |
| Agno | **Approach 1** Supports [configuring an agent](https://docs.agno.com/input-output/structured-output/agent) at agent construction time |
| A2A | **Informal Approach 2** Doesn't formally support schema negotiation, but [hints can be provided via metadata](https://a2a-protocol.org/latest/specification/#97-structured-data-exchange-requesting-and-providing-json) at invocation time |
| Protocol Activity | Supports returning [Complex types](https://github.com/microsoft/Agents/blob/main/specs/activity/protocol-activity.md#complex-types) but no support for requesting a type |
+8
View File
@@ -160,6 +160,14 @@ Sequential orchestration uses a few small adapter nodes for plumbing:
These may appear in event streams (executor_invoked/executor_completed). They're analogous to
concurrents dispatcher and aggregator and can be ignored if you only care about agent activity.
### AzureOpenAIResponsesClient vs AzureAIAgent
Workflow and orchestration samples use `AzureOpenAIResponsesClient` rather than the CRUD-style `AzureAIAgent` client. The key difference:
- **`AzureOpenAIResponsesClient`** — A lightweight client that uses the underlying Agent Service V2 (Responses API) for non-CRUD-style agents. Orchestrations use this client because agents are created locally and do not require server-side lifecycle management (create/update/delete). This is the recommended client for orchestration patterns (Sequential, Concurrent, Handoff, GroupChat, Magentic).
- **`AzureAIAgent`** — A CRUD-style client for server-managed agents. Use this when you need persistent, server-side agent definitions with features like file search, code interpreter sessions, or thread management provided by the Azure AI Agent Service.
### Environment Variables
Workflow samples that use `AzureOpenAIResponsesClient` expect:
@@ -18,10 +18,11 @@ Run with:
"""
import asyncio
import os
from pathlib import Path
from typing import Any
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.declarative import WorkflowFactory
from azure.identity import AzureCliCredential
from pydantic import BaseModel, Field
@@ -196,8 +197,12 @@ def format_order_confirmation(order_data: dict[str, Any], order_calculation: dic
async def main():
"""Run the agent to function tool workflow."""
# Create Azure OpenAI client
chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Create Azure OpenAI Responses client
chat_client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create the order analysis agent with structured output
order_analysis_agent = chat_client.as_agent(
@@ -6,7 +6,7 @@ This sample demonstrates an agent with function tools responding to user queries
The workflow showcases:
- **Function Tools**: Agent equipped with tools to query menu data
- **Real Azure OpenAI Agent**: Uses `AzureOpenAIChatClient` to create an agent with tools
- **Real Azure OpenAI Agent**: Uses `AzureOpenAIResponsesClient` to create an agent with tools
- **Agent Registration**: Shows how to register agents with the `WorkflowFactory`
## Tools
@@ -72,7 +72,11 @@ Session Complete
```python
# Create the agent with tools
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
menu_agent = client.as_agent(
name="MenuAgent",
instructions="You are a helpful restaurant menu assistant...",
@@ -92,6 +92,10 @@ from agent_framework.orchestrations import (
These may appear in event streams (executor_invoked/executor_completed). They're analogous to concurrent's dispatcher and aggregator and can be ignored if you only care about agent activity.
## Why AzureOpenAIResponsesClient?
Orchestration samples use `AzureOpenAIResponsesClient` rather than the CRUD-style `AzureAIAgent` client. Orchestrations create agents locally and do not require server-side lifecycle management (create/update/delete). `AzureOpenAIResponsesClient` is a lightweight client that uses the underlying Agent Service V2 (Responses API) for non-CRUD-style agents, which is ideal for orchestration patterns like Sequential, Concurrent, Handoff, GroupChat, and Magentic.
## Environment Variables
Orchestration samples that use `AzureOpenAIResponsesClient` expect:
@@ -1,10 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import Message
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import ConcurrentBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -26,14 +27,20 @@ Demonstrates:
- Workflow completion when idle with no pending work
Prerequisites:
- Azure OpenAI access configured for AzureOpenAIChatClient (use az login + env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
- Familiarity with Workflow events (WorkflowEvent)
"""
async def main() -> None:
# 1) Create three domain agents using AzureOpenAIChatClient
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# 1) Create three domain agents using AzureOpenAIResponsesClient
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher = client.as_agent(
instructions=(
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import (
@@ -12,7 +13,7 @@ from agent_framework import (
WorkflowContext,
handler,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import ConcurrentBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -29,21 +30,23 @@ and emit AgentExecutorResponse outputs, which allows reuse of the high-level
ConcurrentBuilder API and the default aggregator.
Demonstrates:
- Executors that create their Agent in __init__ (via AzureOpenAIChatClient)
- Executors that create their Agent in __init__ (via AzureOpenAIResponsesClient)
- A @handler that converts AgentExecutorRequest -> AgentExecutorResponse
- ConcurrentBuilder(participants=[...]) to build fan-out/fan-in
- Default aggregator returning list[Message] (one user + one assistant per agent)
- Workflow completion when all participants become idle
Prerequisites:
- Azure OpenAI configured for AzureOpenAIChatClient (az login + required env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
class ResearcherExec(Executor):
agent: Agent
def __init__(self, client: AzureOpenAIChatClient, id: str = "researcher"):
def __init__(self, client: AzureOpenAIResponsesClient, id: str = "researcher"):
self.agent = client.as_agent(
instructions=(
"You're an expert market and product researcher. Given a prompt, provide concise, factual insights,"
@@ -63,7 +66,7 @@ class ResearcherExec(Executor):
class MarketerExec(Executor):
agent: Agent
def __init__(self, client: AzureOpenAIChatClient, id: str = "marketer"):
def __init__(self, client: AzureOpenAIResponsesClient, id: str = "marketer"):
self.agent = client.as_agent(
instructions=(
"You're a creative marketing strategist. Craft compelling value propositions and target messaging"
@@ -83,7 +86,7 @@ class MarketerExec(Executor):
class LegalExec(Executor):
agent: Agent
def __init__(self, client: AzureOpenAIChatClient, id: str = "legal"):
def __init__(self, client: AzureOpenAIResponsesClient, id: str = "legal"):
self.agent = client.as_agent(
instructions=(
"You're a cautious legal/compliance reviewer. Highlight constraints, disclaimers, and policy concerns"
@@ -101,7 +104,11 @@ class LegalExec(Executor):
async def main() -> None:
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher = ResearcherExec(client)
marketer = MarketerExec(client)
@@ -1,10 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import Message
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import ConcurrentBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -17,7 +18,7 @@ Sample: Concurrent Orchestration with Custom Aggregator
Build a concurrent workflow with ConcurrentBuilder that fans out one prompt to
multiple domain agents and fans in their responses. Override the default
aggregator with a custom async callback that uses AzureOpenAIChatClient.get_response()
aggregator with a custom async callback that uses AzureOpenAIResponsesClient.get_response()
to synthesize a concise, consolidated summary from the experts' outputs.
The workflow completes when all participants become idle.
@@ -28,12 +29,18 @@ Demonstrates:
- Workflow output yielded with the synthesized summary string
Prerequisites:
- Azure OpenAI configured for AzureOpenAIChatClient (az login + required env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
async def main() -> None:
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher = client.as_agent(
instructions=(
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import cast
from agent_framework import (
@@ -8,7 +9,7 @@ from agent_framework import (
AgentResponseUpdate,
Message,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -25,7 +26,9 @@ What it does:
- Coordinates a researcher and writer agent to solve tasks collaboratively
Prerequisites:
- OpenAI environment variables configured for OpenAIChatClient
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
ORCHESTRATOR_AGENT_INSTRUCTIONS = """
@@ -39,8 +42,12 @@ Guidelines:
async def main() -> None:
# Create a chat client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Create a Responses client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Orchestrator agent that manages the conversation
# Note: This agent (and the underlying chat client) must support structured outputs.
@@ -2,6 +2,7 @@
import asyncio
import logging
import os
from typing import cast
from agent_framework import (
@@ -9,7 +10,7 @@ from agent_framework import (
AgentResponseUpdate,
Message,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -38,15 +39,21 @@ Participants represent:
- Doctor from Scandinavia (public health, equity, societal support)
Prerequisites:
- OpenAI environment variables configured for OpenAIChatClient
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
# Load environment variables from .env file
load_dotenv()
def _get_chat_client() -> AzureOpenAIChatClient:
return AzureOpenAIChatClient(credential=AzureCliCredential())
def _get_chat_client() -> AzureOpenAIResponsesClient:
return AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
async def main() -> None:
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import cast
from agent_framework import (
@@ -8,7 +9,7 @@ from agent_framework import (
AgentResponseUpdate,
Message,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatBuilder, GroupChatState
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -24,7 +25,9 @@ What it does:
- Uses a pure Python function to control speaker selection based on conversation state
Prerequisites:
- OpenAI environment variables configured for OpenAIChatClient
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
@@ -36,8 +39,12 @@ def round_robin_selector(state: GroupChatState) -> str:
async def main() -> None:
# Create a chat client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Create a Responses client using Azure OpenAI and Azure CLI credentials for all agents
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Participant agents
expert = Agent(
@@ -2,6 +2,7 @@
import asyncio
import logging
import os
from typing import cast
from agent_framework import (
@@ -10,7 +11,7 @@ from agent_framework import (
Message,
resolve_agent_id,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import HandoffBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -28,8 +29,9 @@ Routing Pattern:
User -> Coordinator -> Specialist (iterates N times) -> Handoff -> Final Output
Prerequisites:
- `az login` (Azure CLI authentication)
- Environment variables for AzureOpenAIChatClient (AZURE_OPENAI_ENDPOINT, etc.)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run `az login` before executing the sample.
Key Concepts:
- Autonomous interaction mode: agents iterate until they handoff
@@ -41,7 +43,7 @@ load_dotenv()
def create_agents(
client: AzureOpenAIChatClient,
client: AzureOpenAIResponsesClient,
) -> tuple[Agent, Agent, Agent]:
"""Create coordinator and specialists for autonomous iteration."""
coordinator = client.as_agent(
@@ -77,7 +79,11 @@ def create_agents(
async def main() -> None:
"""Run an autonomous handoff workflow with specialist iteration enabled."""
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
coordinator, research_agent, summary_agent = create_agents(client)
# Build the workflow with autonomous mode
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Annotated, cast
from agent_framework import (
@@ -11,7 +12,7 @@ from agent_framework import (
WorkflowRunState,
tool,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import HandoffAgentUserRequest, HandoffBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -25,8 +26,9 @@ A handoff workflow defines a pattern that assembles agents in a mesh topology, a
them to transfer control to each other based on the conversation context.
Prerequisites:
- `az login` (Azure CLI authentication)
- Environment variables configured for AzureOpenAIChatClient (AZURE_OPENAI_ENDPOINT, etc.)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run `az login` before executing the sample.
Key Concepts:
- Auto-registered handoff tools: HandoffBuilder automatically creates handoff tools
@@ -58,11 +60,11 @@ def process_return(order_number: Annotated[str, "Order number to process return
return f"Return initiated successfully for order {order_number}. You will receive return instructions via email."
def create_agents(client: AzureOpenAIChatClient) -> tuple[Agent, Agent, Agent, Agent]:
def create_agents(client: AzureOpenAIResponsesClient) -> tuple[Agent, Agent, Agent, Agent]:
"""Create and configure the triage and specialist agents.
Args:
client: The AzureOpenAIChatClient to use for creating agents.
client: The AzureOpenAIResponsesClient to use for creating agents.
Returns:
Tuple of (triage_agent, refund_agent, order_agent, return_agent)
@@ -192,8 +194,12 @@ async def main() -> None:
the demo reproducible and testable. In a production application, you would
replace the scripted_responses with actual user input collection.
"""
# Initialize the Azure OpenAI chat client
client = AzureOpenAIChatClient(credential=AzureCliCredential())
# Initialize the Azure OpenAI Responses client
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create all agents: triage + specialists
triage, refund, order, support = create_agents(client)
@@ -3,6 +3,7 @@
import asyncio
import json
import logging
import os
from typing import cast
from agent_framework import (
@@ -11,8 +12,9 @@ from agent_framework import (
Message,
WorkflowEvent,
)
from agent_framework.openai import OpenAIChatClient, OpenAIResponsesClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import GroupChatRequestSentEvent, MagenticBuilder, MagenticProgressLedger
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
logging.basicConfig(level=logging.WARNING)
@@ -40,7 +42,9 @@ energy efficiency and CO2 emissions of several ML models, streams intermediate
events, and prints the final answer. The workflow completes when idle.
Prerequisites:
- OpenAI credentials configured for `OpenAIChatClient` and `OpenAIResponsesClient`.
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
# Load environment variables from .env file
@@ -48,25 +52,29 @@ load_dotenv()
async def main() -> None:
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher_agent = Agent(
name="ResearcherAgent",
description="Specialist in research and information gathering",
instructions=(
"You are a Researcher. You find information without additional computation or quantitative analysis."
),
# This agent requires the gpt-4o-search-preview model to perform web searches.
client=OpenAIChatClient(model_id="gpt-4o-search-preview"),
client=client,
)
# Create code interpreter tool using instance method
coder_client = OpenAIResponsesClient()
code_interpreter_tool = coder_client.get_code_interpreter_tool()
code_interpreter_tool = client.get_code_interpreter_tool()
coder_agent = Agent(
name="CoderAgent",
description="A helpful assistant that writes and executes code to process and analyze data.",
instructions="You solve questions using code. Please provide detailed analysis and computation process.",
client=coder_client,
client=client,
tools=code_interpreter_tool,
)
@@ -75,7 +83,7 @@ async def main() -> None:
name="MagenticManager",
description="Orchestrator that coordinates the research and coding workflow",
instructions="You coordinate a team to complete complex tasks efficiently.",
client=OpenAIChatClient(),
client=client,
)
print("\nBuilding Magentic Workflow...")
@@ -2,6 +2,7 @@
import asyncio
import json
import os
from datetime import datetime
from pathlib import Path
from typing import cast
@@ -14,9 +15,9 @@ from agent_framework import (
WorkflowEvent,
WorkflowRunState,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import MagenticBuilder, MagenticPlanReviewRequest
from azure.identity._credentials import AzureCliCredential
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
# Load environment variables from .env file
@@ -38,7 +39,9 @@ Concepts highlighted here:
`responses` mapping so we can inject the stored human reply during restoration.
Prerequisites:
- OpenAI environment variables configured for `OpenAIChatClient`.
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
TASK = (
@@ -61,14 +64,22 @@ def build_workflow(checkpoint_storage: FileCheckpointStorage):
name="ResearcherAgent",
description="Collects background facts and references for the project.",
instructions=("You are the research lead. Gather crisp bullet points the team should know."),
client=AzureOpenAIChatClient(credential=AzureCliCredential()),
client=AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
),
)
writer = Agent(
name="WriterAgent",
description="Synthesizes the final brief for stakeholders.",
instructions=("You convert the research notes into a structured brief with milestones and risks."),
client=AzureOpenAIChatClient(credential=AzureCliCredential()),
client=AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
),
)
# Create a manager agent for orchestration
@@ -76,7 +87,11 @@ def build_workflow(checkpoint_storage: FileCheckpointStorage):
name="MagenticManager",
description="Orchestrator that coordinates the research and writing workflow",
instructions="You coordinate a team to complete complex tasks efficiently.",
client=AzureOpenAIChatClient(credential=AzureCliCredential()),
client=AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
),
)
# The builder wires in the Magentic orchestrator, sets the plan review path, and
@@ -2,6 +2,7 @@
import asyncio
import json
import os
from collections.abc import AsyncIterable
from typing import cast
@@ -11,8 +12,9 @@ from agent_framework import (
Message,
WorkflowEvent,
)
from agent_framework.openai import OpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import MagenticBuilder, MagenticPlanReviewRequest, MagenticPlanReviewResponse
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
# Load environment variables from .env file
@@ -35,7 +37,9 @@ Plan review options:
- revise(feedback): Provide textual feedback to modify the plan
Prerequisites:
- OpenAI credentials configured for `OpenAIChatClient`.
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
# Keep track of the last response to format output nicely in streaming mode
@@ -96,25 +100,31 @@ async def process_event_stream(stream: AsyncIterable[WorkflowEvent]) -> dict[str
async def main() -> None:
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
researcher_agent = Agent(
name="ResearcherAgent",
description="Specialist in research and information gathering",
instructions="You are a Researcher. You find information and gather facts.",
client=OpenAIChatClient(model_id="gpt-4o"),
client=client,
)
analyst_agent = Agent(
name="AnalystAgent",
description="Data analyst who processes and summarizes research findings",
instructions="You are an Analyst. You analyze findings and create summaries.",
client=OpenAIChatClient(model_id="gpt-4o"),
client=client,
)
manager_agent = Agent(
name="MagenticManager",
description="Orchestrator that coordinates the workflow",
instructions="You coordinate a team to complete tasks efficiently.",
client=OpenAIChatClient(model_id="gpt-4o"),
client=client,
)
print("\nBuilding Magentic Workflow with Human Plan Review...")
@@ -1,10 +1,11 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import cast
from agent_framework import Message
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -28,13 +29,19 @@ Note on internal adapters:
You can safely ignore them when focusing on agent progress.
Prerequisites:
- Azure OpenAI access configured for AzureOpenAIChatClient (use az login + env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
async def main() -> None:
# 1) Create agents
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
writer = client.as_agent(
instructions=("You are a concise copywriter. Provide a single, punchy marketing sentence based on the prompt."),
@@ -1,6 +1,7 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from typing import Any
from agent_framework import (
@@ -10,7 +11,7 @@ from agent_framework import (
WorkflowContext,
handler,
)
from agent_framework.azure import AzureOpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.orchestrations import SequentialBuilder
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
@@ -32,7 +33,9 @@ Custom executor contract:
- Emit the updated conversation via ctx.send_message([...])
Prerequisites:
- Azure OpenAI access configured for AzureOpenAIChatClient (use az login + env vars)
- AZURE_AI_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- Azure OpenAI configured for AzureOpenAIResponsesClient with required environment variables.
- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample.
"""
@@ -62,7 +65,11 @@ class Summarizer(Executor):
async def main() -> None:
# 1) Create a content agent
client = AzureOpenAIChatClient(credential=AzureCliCredential())
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
content = client.as_agent(
instructions="Produce a concise paragraph answering the user's request.",
name="content",