agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: Fix tool normalization and provider sample consolidation (#3953)

Browse Source 
* Fix tool normalization and provider samples

- restore callable/single-tool normalization paths and unset tool-choice behavior\n- consolidate and expand chat/provider samples (OpenAI/Azure/Anthropic/Ollama/Bedrock)\n- migrate Bedrock lazy import surface to agent_framework.amazon and move provider samples

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* small fix in sample

* Finalize provider, samples, and core cleanup

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix CopilotTool passthrough in agent

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix link

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard van Valkenburg
2026-02-16 17:30:38 +01:00
committed by GitHub
Unverified
parent ed113f941c
commit aab621f5eb
99 changed files with 1190 additions and 969 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/01-get-started/01_hello_agent.py
+2
View File
@@ -12,6 +12,8 @@ Hello Agent — Simplest possible agent
This sample creates a minimal agent using AzureOpenAIResponsesClient via an
Azure AI Foundry project endpoint, and runs it in both non-streaming and streaming modes.
There are XML tags in all of the get started samples, those are used to display the same code in the docs repo.
Environment variables:
AZURE_AI_PROJECT_ENDPOINT — Your Azure AI Foundry project endpoint
AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME — Model deployment name (e.g. gpt-4o)
python/samples/01-get-started/README.md
+4 -2
View File
@@ -12,8 +12,8 @@ pip install agent-framework --pre
Set the required environment variables:
```bash
export OPENAI_API_KEY="sk-..."
export OPENAI_RESPONSES_MODEL_ID="gpt-4o" # optional, defaults to gpt-4o
export AZURE_AI_PROJECT_ENDPOINT="https://your-project-endpoint"
export AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME="gpt-4o" # optional, defaults to gpt-4o
```
## Samples
@@ -32,3 +32,5 @@ Run any sample with:
```bash
python 01_hello_agent.py
```
These samples use Azure Foundry models with the Responses API. To switch providers, just replace the client, see [all providers](../02-agents/providers/README.md)
python/samples/02-agents/chat_client/README.md
+50 -17
View File
@@ -1,41 +1,74 @@
# Chat Client Examples
This folder contains simple examples demonstrating direct usage of various chat clients.
This folder contains examples for direct chat client usage patterns.
## Examples
| File | Description |
|------|-------------|
| [`azure_assistants_client.py`](azure_assistants_client.py) | Direct usage of Azure Assistants Client for basic chat interactions with Azure OpenAI assistants. |
| [`azure_chat_client.py`](azure_chat_client.py) | Direct usage of Azure Chat Client for chat interactions with Azure OpenAI models. |
| [`azure_responses_client.py`](azure_responses_client.py) | Direct usage of Azure Responses Client for structured response generation with Azure OpenAI models. |
| [`built_in_chat_clients.py`](built_in_chat_clients.py) | Consolidated sample for built-in chat clients. Uses `get_client()` to create the selected client and pass it to `main()`. |
| [`chat_response_cancellation.py`](chat_response_cancellation.py) | Demonstrates how to cancel chat responses during streaming, showing proper cancellation handling and cleanup. |
| [`azure_ai_chat_client.py`](azure_ai_chat_client.py) | Direct usage of Azure AI Chat Client for chat interactions with Azure AI models. |
| [`openai_assistants_client.py`](openai_assistants_client.py) | Direct usage of OpenAI Assistants Client for basic chat interactions with OpenAI assistants. |
| [`openai_chat_client.py`](openai_chat_client.py) | Direct usage of OpenAI Chat Client for chat interactions with OpenAI models. |
| [`openai_responses_client.py`](openai_responses_client.py) | Direct usage of OpenAI Responses Client for structured response generation with OpenAI models. |
| [`custom_chat_client.py`](custom_chat_client.py) | Demonstrates how to create custom chat clients by extending the `BaseChatClient` class. Shows a `EchoingChatClient` implementation and how to integrate it with `Agent` using the `as_agent()` method. |
## Selecting a built-in client
`built_in_chat_clients.py` starts with:
```python
asyncio.run(main("openai_chat"))
```
Change the argument to pick a client:
- `openai_chat`
- `openai_responses`
- `openai_assistants`
- `anthropic`
- `ollama`
- `bedrock`
- `azure_openai_chat`
- `azure_openai_responses`
- `azure_openai_responses_foundry`
- `azure_openai_assistants`
- `azure_ai_agent`
Example:
```bash
uv run samples/02-agents/chat_client/built_in_chat_clients.py
```
## Environment Variables
Depending on which client you're using, set the appropriate environment variables:
Depending on the selected client, set the appropriate environment variables:
**For Azure clients:**
- `AZURE_OPENAI_ENDPOINT`: Your Azure OpenAI endpoint
- `AZURE_OPENAI_CHAT_DEPLOYMENT_NAME`: The name of your Azure OpenAI chat deployment
- `AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME`: The name of your Azure OpenAI responses deployment
**For Azure AI client:**
**For Azure OpenAI Foundry responses client (`azure_openai_responses_foundry`):**
- `AZURE_AI_PROJECT_ENDPOINT`: Your Azure AI project endpoint
- `AZURE_AI_MODEL_DEPLOYMENT_NAME`: The name of your model deployment
- `AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME`: The name of your Azure OpenAI responses deployment
**For Azure AI agent client (`azure_ai_agent`):**
- `AZURE_AI_PROJECT_ENDPOINT`: Your Azure AI project endpoint
- `AZURE_AI_MODEL_DEPLOYMENT_NAME`: The name of your model deployment (used by `azure_ai_agent`)
**For OpenAI clients:**
- `OPENAI_API_KEY`: Your OpenAI API key
- `OPENAI_CHAT_MODEL_ID`: The OpenAI model to use for chat clients (e.g., `gpt-4o`, `gpt-4o-mini`, `gpt-3.5-turbo`)
- `OPENAI_RESPONSES_MODEL_ID`: The OpenAI model to use for responses clients (e.g., `gpt-4o`, `gpt-4o-mini`, `gpt-3.5-turbo`)
- `OPENAI_CHAT_MODEL_ID`: The OpenAI model for `openai_chat` and `openai_assistants`
- `OPENAI_RESPONSES_MODEL_ID`: The OpenAI model for `openai_responses`
**For Ollama client:**
- `OLLAMA_HOST`: Your Ollama server URL (defaults to `http://localhost:11434` if not set)
- `OLLAMA_MODEL_ID`: The Ollama model to use for chat (e.g., `llama3.2`, `llama2`, `codellama`)
**For Anthropic client (`anthropic`):**
- `ANTHROPIC_API_KEY`: Your Anthropic API key
- `ANTHROPIC_CHAT_MODEL_ID`: The Anthropic model ID (for example, `claude-sonnet-4-5`)
> **Note**: For Ollama, ensure you have Ollama installed and running locally with at least one model downloaded. Visit [https://ollama.com/](https://ollama.com/) for installation instructions.
**For Ollama client (`ollama`):**
- `OLLAMA_HOST`: Ollama server URL (defaults to `http://localhost:11434` if unset)
- `OLLAMA_MODEL_ID`: Ollama model name (for example, `mistral`, `qwen2.5:8b`)
**For Bedrock client (`bedrock`):**
- `BEDROCK_CHAT_MODEL_ID`: Bedrock model ID (for example, `anthropic.claude-3-5-sonnet-20240620-v1:0`)
- `BEDROCK_REGION`: AWS region (defaults to `us-east-1` if unset)
- AWS credentials via standard environment variables (for example, `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`)
python/samples/02-agents/chat_client/azure_ai_chat_client.py
-49
View File
@@ -1,49 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field
"""
Azure AI Chat Client Direct Usage Example
Demonstrates direct AzureAIChatClient usage for chat interactions with Azure AI models.
Shows function calling capabilities with custom business logic.
"""
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def main() -> None:
# For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
# authentication option.
async with AzureAIAgentClient(credential=AzureCliCredential()) as client:
message = "What's the weather in Amsterdam and in Paris?"
stream = False
print(f"User: {message}")
if stream:
print("Assistant: ", end="")
async for chunk in client.get_response(message, tools=get_weather, stream=True):
if str(chunk):
print(str(chunk), end="")
print("")
else:
response = await client.get_response(message, tools=get_weather)
print(f"Assistant: {response}")
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/chat_client/azure_assistants_client.py
-49
View File
@@ -1,49 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework.azure import AzureOpenAIAssistantsClient
from azure.identity import AzureCliCredential
from pydantic import Field
"""
Azure Assistants Client Direct Usage Example
Demonstrates direct AzureAssistantsClient usage for chat interactions with Azure OpenAI assistants.
Shows function calling capabilities and automatic assistant creation.
"""
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def main() -> None:
# For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
# authentication option.
async with AzureOpenAIAssistantsClient(credential=AzureCliCredential()) as client:
message = "What's the weather in Amsterdam and in Paris?"
stream = False
print(f"User: {message}")
if stream:
print("Assistant: ", end="")
async for chunk in client.get_response(message, tools=get_weather, stream=True):
if str(chunk):
print(str(chunk), end="")
print("")
else:
response = await client.get_response(message, tools=get_weather)
print(f"Assistant: {response}")
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/chat_client/azure_chat_client.py
-49
View File
@@ -1,49 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential
from pydantic import Field
"""
Azure Chat Client Direct Usage Example
Demonstrates direct AzureChatClient usage for chat interactions with Azure OpenAI models.
Shows function calling capabilities with custom business logic.
"""
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def main() -> None:
# For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
# authentication option.
client = AzureOpenAIChatClient(credential=AzureCliCredential())
message = "What's the weather in Amsterdam and in Paris?"
stream = False
print(f"User: {message}")
if stream:
print("Assistant: ", end="")
async for chunk in client.get_response(message, tools=get_weather, stream=True):
if str(chunk):
print(str(chunk), end="")
print("")
else:
response = await client.get_response(message, tools=get_weather)
print(f"Assistant: {response}")
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/chat_client/azure_responses_client.py
-95
View File
@@ -1,95 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
from pydantic import BaseModel
"""
Azure Responses Client Direct Usage Example
Demonstrates direct AzureResponsesClient usage for structured response generation with Azure OpenAI models.
Shows function calling capabilities with custom business logic.
"""
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, "The location to get the weather for."],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
@tool(approval_mode="never_require")
def get_time():
"""Get the current time."""
from datetime import datetime
now = datetime.now()
return f"The current date time is {now.strftime('%Y-%m-%d - %H:%M:%S')}."
class WeatherDetail(BaseModel):
"""Structured output for weather information."""
location: str
weather: str
class Weather(BaseModel):
"""Container for multiple outputs."""
date_time: str
weather_details: list[WeatherDetail]
async def main() -> None:
# For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
# authentication option.
client = AzureOpenAIResponsesClient(credential=AzureCliCredential(), api_version="preview")
message = "What's the weather in Amsterdam and in Paris?"
stream = True
print(f"User: {message}")
response = client.get_response(
message,
options={"response_format": Weather, "tools": [get_weather, get_time]},
stream=stream,
)
if stream:
response = await response.get_final_response()
else:
response = await response
if result := response.value:
print(f"Assistant: {result.model_dump_json(indent=2)}")
else:
print(f"Assistant: {response.text}")
# Expected output (time will be different):
"""
User: What's the weather in Amsterdam and in Paris?
Assistant: {
"date_time": "2026-02-06 - 13:30:40",
"weather_details": [
{
"location": "Amsterdam",
"weather": "The weather in Amsterdam is cloudy with a high of 21°C."
},
{
"location": "Paris",
"weather": "The weather in Paris is sunny with a high of 27°C."
}
]
}
"""
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/chat_client/built_in_chat_clients.py
+156
View File
@@ -0,0 +1,156 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from random import randint
from typing import Annotated, Any, Literal
from agent_framework import SupportsChatGetResponse, tool
from agent_framework.azure import (
AzureAIAgentClient,
AzureOpenAIAssistantsClient,
)
from agent_framework.openai import OpenAIAssistantsClient
from azure.identity import AzureCliCredential
from azure.identity.aio import AzureCliCredential as AsyncAzureCliCredential
from pydantic import Field
"""
Built-in Chat Clients Example
This sample demonstrates how to run the same prompt flow against different built-in
chat clients using a single `get_client` factory.
Select one of these client names:
- openai_chat
- openai_responses
- openai_assistants
- anthropic
- ollama
- bedrock
- azure_openai_chat
- azure_openai_responses
- azure_openai_responses_foundry
- azure_openai_assistants
- azure_ai_agent
"""
ClientName = Literal[
"openai_chat",
"openai_responses",
"openai_assistants",
"anthropic",
"ollama",
"bedrock",
"azure_openai_chat",
"azure_openai_responses",
"azure_openai_responses_foundry",
"azure_openai_assistants",
"azure_ai_agent",
]
# NOTE: approval_mode="never_require" is for sample brevity.
# Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py
# and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
def get_client(client_name: ClientName) -> SupportsChatGetResponse[Any]:
"""Create a built-in chat client from a name."""
from agent_framework.amazon import BedrockChatClient
from agent_framework.anthropic import AnthropicClient
from agent_framework.azure import (
AzureOpenAIChatClient,
AzureOpenAIResponsesClient,
)
from agent_framework.ollama import OllamaChatClient
from agent_framework.openai import OpenAIChatClient, OpenAIResponsesClient
# 1. Create OpenAI clients.
if client_name == "openai_chat":
return OpenAIChatClient()
if client_name == "openai_responses":
return OpenAIResponsesClient()
if client_name == "openai_assistants":
return OpenAIAssistantsClient()
if client_name == "anthropic":
return AnthropicClient()
if client_name == "ollama":
return OllamaChatClient()
if client_name == "bedrock":
return BedrockChatClient()
# 2. Create Azure OpenAI clients.
if client_name == "azure_openai_chat":
return AzureOpenAIChatClient(credential=AzureCliCredential())
if client_name == "azure_openai_responses":
return AzureOpenAIResponsesClient(credential=AzureCliCredential(), api_version="preview")
if client_name == "azure_openai_responses_foundry":
return AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
if client_name == "azure_openai_assistants":
return AzureOpenAIAssistantsClient(credential=AzureCliCredential())
# 3. Create Azure AI client.
if client_name == "azure_ai_agent":
return AzureAIAgentClient(credential=AsyncAzureCliCredential())
raise ValueError(f"Unsupported client name: {client_name}")
async def main(client_name: ClientName = "openai_chat") -> None:
"""Run a basic prompt using a selected built-in client."""
client = get_client(client_name)
# 1. Configure prompt and streaming mode.
message = "What's the weather in Amsterdam and in Paris?"
stream = os.getenv("STREAM", "false").lower() == "true"
print(f"Client: {client_name}")
print(f"User: {message}")
# 2. Run with context-managed clients.
if isinstance(client, OpenAIAssistantsClient | AzureOpenAIAssistantsClient | AzureAIAgentClient):
async with client:
if stream:
response_stream = client.get_response(message, stream=True, options={"tools": get_weather})
print("Assistant: ", end="")
async for chunk in response_stream:
if chunk.text:
print(chunk.text, end="")
print("")
else:
print(f"Assistant: {await client.get_response(message, stream=False, options={'tools': get_weather})}")
return
# 3. Run with non-context-managed clients.
if stream:
response_stream = client.get_response(message, stream=True, options={"tools": get_weather})
print("Assistant: ", end="")
async for chunk in response_stream:
if chunk.text:
print(chunk.text, end="")
print("")
else:
print(f"Assistant: {await client.get_response(message, stream=False, options={'tools': get_weather})}")
if __name__ == "__main__":
asyncio.run(main("openai_chat"))
"""
Sample output:
User: What's the weather in Amsterdam and in Paris?
Assistant: The weather in Amsterdam is sunny with a high of 25°C.
...and in Paris it is cloudy with a high of 19°C.
"""
python/samples/02-agents/chat_client/custom_chat_client.py
+36 -19
View File
@@ -4,7 +4,7 @@ import asyncio
import random
import sys
from collections.abc import AsyncIterable, Awaitable, Mapping, Sequence
from typing import Any, ClassVar, Generic
from typing import Any, ClassVar, TypeAlias, TypedDict
from agent_framework import (
BaseChatClient,
@@ -15,15 +15,9 @@ from agent_framework import (
FunctionInvocationLayer,
Message,
ResponseStream,
Role,
)
from agent_framework._clients import OptionsCoT
from agent_framework.observability import ChatTelemetryLayer
if sys.version_info >= (3, 13):
pass
else:
pass
if sys.version_info >= (3, 12):
from typing import override # type: ignore # pragma: no cover
else:
@@ -38,7 +32,18 @@ middleware, telemetry, and function invocation layers explicitly.
"""
class EchoingChatClient(BaseChatClient[OptionsCoT], Generic[OptionsCoT]):
class EchoingChatClientOptions(TypedDict, total=False):
"""Custom options for EchoingChatClient."""
uppercase: bool
suffix: str
stream_delay_seconds: float
OptionsT: TypeAlias = EchoingChatClientOptions
class EchoingChatClient(BaseChatClient[OptionsT]):
"""A custom chat client that echoes messages back with modifications.
This demonstrates how to implement a custom chat client by extending BaseChatClient
@@ -73,7 +78,7 @@ class EchoingChatClient(BaseChatClient[OptionsCoT], Generic[OptionsCoT]):
# Echo the last user message
last_user_message = None
for message in reversed(messages):
if message.role == Role.USER:
if message.role == "user":
last_user_message = message
break
@@ -82,7 +87,13 @@ class EchoingChatClient(BaseChatClient[OptionsCoT], Generic[OptionsCoT]):
else:
response_text = f"{self.prefix} [No text message found]"
response_message = Message(role=Role.ASSISTANT, contents=[Content.from_text(response_text)])
if options.get("uppercase"):
response_text = response_text.upper()
if suffix := options.get("suffix"):
response_text = f"{response_text} {suffix}"
stream_delay_seconds = float(options.get("stream_delay_seconds", 0.05))
response_message = Message(role="assistant", contents=[Content.from_text(response_text)])
response = ChatResponse(
messages=[response_message],
@@ -102,21 +113,20 @@ class EchoingChatClient(BaseChatClient[OptionsCoT], Generic[OptionsCoT]):
for char in response_text_local:
yield ChatResponseUpdate(
contents=[Content.from_text(char)],
role=Role.ASSISTANT,
role="assistant",
response_id=f"echo-stream-resp-{random.randint(1000, 9999)}",
model_id="echo-model-v1",
)
await asyncio.sleep(0.05)
await asyncio.sleep(stream_delay_seconds)
return ResponseStream(_stream(), finalizer=lambda updates: response)
class EchoingChatClientWithLayers( # type: ignore[misc,type-var]
ChatMiddlewareLayer[OptionsCoT],
ChatTelemetryLayer[OptionsCoT],
FunctionInvocationLayer[OptionsCoT],
EchoingChatClient[OptionsCoT],
Generic[OptionsCoT],
class EchoingChatClientWithLayers( # type: ignore[misc]
ChatMiddlewareLayer[OptionsT],
ChatTelemetryLayer[OptionsT],
FunctionInvocationLayer[OptionsT],
EchoingChatClient,
):
"""Echoing chat client that explicitly composes middleware, telemetry, and function layers."""
@@ -134,7 +144,14 @@ async def main() -> None:
# Use the chat client directly
print("Using chat client directly:")
direct_response = await echo_client.get_response("Hello, custom chat client!")
direct_response = await echo_client.get_response(
"Hello, custom chat client!",
options={
"uppercase": True,
"suffix": "(CUSTOM OPTIONS)",
"stream_delay_seconds": 0.02,
},
)
print(f"Direct response: {direct_response.messages[0].text}")
# Create an agent using the custom chat client
python/samples/02-agents/chat_client/openai_assistants_client.py
-47
View File
@@ -1,47 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework.openai import OpenAIAssistantsClient
from pydantic import Field
"""
OpenAI Assistants Client Direct Usage Example
Demonstrates direct OpenAIAssistantsClient usage for chat interactions with OpenAI assistants.
Shows function calling capabilities and automatic assistant creation.
"""
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def main() -> None:
async with OpenAIAssistantsClient() as client:
message = "What's the weather in Amsterdam and in Paris?"
stream = False
print(f"User: {message}")
if stream:
print("Assistant: ", end="")
async for chunk in client.get_response(message, tools=get_weather, stream=True):
if str(chunk):
print(str(chunk), end="")
print("")
else:
response = await client.get_response(message, tools=get_weather)
print(f"Assistant: {response}")
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/chat_client/openai_chat_client.py
-47
View File
@@ -1,47 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework.openai import OpenAIChatClient
from pydantic import Field
"""
OpenAI Chat Client Direct Usage Example
Demonstrates direct OpenAIChatClient usage for chat interactions with OpenAI models.
Shows function calling capabilities with custom business logic.
"""
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def main() -> None:
client = OpenAIChatClient()
message = "What's the weather in Amsterdam and in Paris?"
stream = True
print(f"User: {message}")
if stream:
print("Assistant: ", end="")
async for chunk in client.get_response(message, tools=get_weather, stream=True):
if chunk.text:
print(chunk.text, end="")
print("")
else:
response = await client.get_response(message, tools=get_weather)
print(f"Assistant: {response}")
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/chat_client/openai_responses_client.py
-47
View File
@@ -1,47 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework.openai import OpenAIResponsesClient
from pydantic import Field
"""
OpenAI Responses Client Direct Usage Example
Demonstrates direct OpenAIResponsesClient usage for structured response generation with OpenAI models.
Shows function calling capabilities with custom business logic.
"""
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def main() -> None:
client = OpenAIResponsesClient()
message = "What's the weather in Amsterdam and in Paris?"
stream = True
print(f"User: {message}")
print("Assistant: ", end="")
response = client.get_response(message, stream=stream, options={"tools": get_weather})
if stream:
# TODO: review names of the methods, could be related to things like HTTP clients?
response.with_transform_hook(lambda chunk: print(chunk.text, end=""))
await response.get_final_response()
else:
response = await response
print(f"Assistant: {response}")
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/context_providers/azure_ai_search/azure_ai_with_search_context_agentic.py
+2
View File
@@ -75,6 +75,7 @@ async def main() -> None:
if knowledge_base_name:
# Use existing Knowledge Base - simplest approach
search_provider = AzureAISearchContextProvider(
source_id="search_provider",
endpoint=search_endpoint,
api_key=search_key,
credential=AzureCliCredential() if not search_key else None,
@@ -91,6 +92,7 @@ async def main() -> None:
if not azure_openai_resource_url:
raise ValueError("AZURE_OPENAI_RESOURCE_URL required when using index_name")
search_provider = AzureAISearchContextProvider(
source_id="search_provider",
endpoint=search_endpoint,
index_name=index_name,
api_key=search_key,
python/samples/02-agents/context_providers/azure_ai_search/azure_ai_with_search_context_semantic.py
+1
View File
@@ -53,6 +53,7 @@ async def main() -> None:
# Create Azure AI Search context provider with semantic mode (recommended, fast)
print("Using SEMANTIC mode (hybrid search + semantic ranking, fast)\n")
search_provider = AzureAISearchContextProvider(
source_id="search_provider",
endpoint=search_endpoint,
index_name=index_name,
api_key=search_key, # Use api_key for API key auth, or credential for managed identity
python/samples/02-agents/context_providers/mem0/mem0_basic.py
+1 -1
View File
@@ -39,7 +39,7 @@ async def main() -> None:
name="FriendlyAssistant",
instructions="You are a friendly assistant.",
tools=retrieve_company_report,
context_providers=[Mem0ContextProvider(user_id=user_id)],
context_providers=[Mem0ContextProvider(source_id="mem0", user_id=user_id)],
) as agent,
):
# First ask the agent to retrieve a company report with no previous context.
python/samples/02-agents/context_providers/mem0/mem0_oss.py
+4 -2
View File
@@ -10,7 +10,9 @@ from azure.identity.aio import AzureCliCredential
from mem0 import AsyncMemory
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
# NOTE: approval_mode="never_require" is for sample brevity.
# Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py
# and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def retrieve_company_report(company_code: str, detailed: bool) -> str:
if company_code != "CNTS":
@@ -42,7 +44,7 @@ async def main() -> None:
name="FriendlyAssistant",
instructions="You are a friendly assistant.",
tools=retrieve_company_report,
context_providers=[Mem0ContextProvider(user_id=user_id, mem0_client=local_mem0_client)],
context_providers=[Mem0ContextProvider(source_id="mem0", user_id=user_id, mem0_client=local_mem0_client)],
) as agent,
):
# First ask the agent to retrieve a company report with no previous context.
python/samples/02-agents/context_providers/mem0/mem0_sessions.py
+27 -15
View File
@@ -34,11 +34,14 @@ async def example_global_thread_scope() -> None:
name="GlobalMemoryAssistant",
instructions="You are an assistant that remembers user preferences across conversations.",
tools=get_user_preferences,
context_providers=[Mem0ContextProvider(
user_id=user_id,
thread_id=global_thread_id,
scope_to_per_operation_thread_id=False, # Share memories across all sessions
)],
context_providers=[
Mem0ContextProvider(
source_id="mem0",
user_id=user_id,
thread_id=global_thread_id,
scope_to_per_operation_thread_id=False, # Share memories across all sessions
)
],
) as global_agent,
):
# Store some preferences in the global scope
@@ -72,10 +75,13 @@ async def example_per_operation_thread_scope() -> None:
name="ScopedMemoryAssistant",
instructions="You are an assistant with thread-scoped memory.",
tools=get_user_preferences,
context_providers=[Mem0ContextProvider(
user_id=user_id,
scope_to_per_operation_thread_id=True, # Isolate memories per session
)],
context_providers=[
Mem0ContextProvider(
source_id="mem0",
user_id=user_id,
scope_to_per_operation_thread_id=True, # Isolate memories per session
)
],
) as scoped_agent,
):
# Create a specific session for this scoped provider
@@ -119,16 +125,22 @@ async def example_multiple_agents() -> None:
AzureAIAgentClient(credential=credential).as_agent(
name="PersonalAssistant",
instructions="You are a personal assistant that helps with personal tasks.",
context_providers=[Mem0ContextProvider(
agent_id=agent_id_1,
)],
context_providers=[
Mem0ContextProvider(
source_id="mem0",
agent_id=agent_id_1,
)
],
) as personal_agent,
AzureAIAgentClient(credential=credential).as_agent(
name="WorkAssistant",
instructions="You are a work assistant that helps with professional tasks.",
context_providers=[Mem0ContextProvider(
agent_id=agent_id_2,
)],
context_providers=[
Mem0ContextProvider(
source_id="mem0",
agent_id=agent_id_2,
)
],
) as work_agent,
):
# Store personal information
python/samples/02-agents/context_providers/redis/README.md
+15 -4
View File
@@ -20,7 +20,8 @@ This folder contains an example demonstrating how to use the Redis context provi
1. A running Redis with RediSearch (Redis Stack or a managed service)
2. Python environment with Agent Framework Redis extra installed
3. Optional: OpenAI API key if using vector embeddings
3. Azure AI Foundry project endpoint and Azure OpenAI Responses deployment
4. Optional: OpenAI API key if using vector embeddings
### Install the package
@@ -50,6 +51,8 @@ See quickstart: `https://learn.microsoft.com/azure/redis/quickstart-create-manag
### Environment variables
- `AZURE_AI_PROJECT_ENDPOINT` (required): Azure AI Foundry project endpoint for `AzureOpenAIResponsesClient`
- `AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME` (required): Azure OpenAI Responses deployment name
- `OPENAI_API_KEY` (optional): Required only if you set `vectorizer_choice="openai"` to enable hybrid search.
### Provider configuration highlights
@@ -70,19 +73,26 @@ The provider supports both full‑text only and hybrid vector search:
2. Agent integration: teaches the agent a preference and verifies it is remembered across turns.
3. Agent + tool: calls a sample tool (flight search) and then asks the agent to recall details remembered from the tool output.
It uses OpenAI for both chat (via `OpenAIChatClient`) and, in some steps, optional embeddings for hybrid search.
It uses `AzureOpenAIResponsesClient` (Foundry project endpoint setup) for chat and, in some steps, optional OpenAI embeddings for hybrid search.
## How to run
1) Start Redis (see options above). For local default, ensure it's reachable at `redis://localhost:6379`.
2) Set your OpenAI key if using embeddings and for the chat client used in the sample:
2) Set Azure Foundry/OpenAI responses environment variables:
```bash
export AZURE_AI_PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>"
export AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME="<deployment-name>"
```
3) (Optional) Set your OpenAI key if using embeddings:
```bash
export OPENAI_API_KEY="<your key>"
```
3) Run the example:
4) Run the example:
```bash
python redis_basics.py
@@ -109,5 +119,6 @@ You should see the agent responses and, when using embeddings, context retrieved
## Troubleshooting
- Ensure at least one of `application_id`, `agent_id`, `user_id`, or `thread_id` is set; the provider requires a scope.
- Verify `AZURE_AI_PROJECT_ENDPOINT` and `AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME` are set for the chat client.
- If using embeddings, verify `OPENAI_API_KEY` is set and reachable.
- Make sure Redis exposes RediSearch (Redis Stack image or managed service with search enabled).
python/samples/02-agents/context_providers/redis/azure_redis_conversation.py
+13 -10
View File
@@ -13,24 +13,25 @@ Requirements:
Environment Variables:
- AZURE_REDIS_HOST: Your Azure Managed Redis host (e.g., myredis.redis.cache.windows.net)
- OPENAI_API_KEY: Your OpenAI API key
- OPENAI_CHAT_MODEL_ID: OpenAI model (e.g., gpt-4o-mini)
- AZURE_AI_PROJECT_ENDPOINT: Your Azure AI Foundry project endpoint
- AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME: Azure OpenAI Responses deployment name
- AZURE_USER_OBJECT_ID: Your Azure AD User Object ID for authentication
"""
import asyncio
import os
from agent_framework.openai import OpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.redis import RedisHistoryProvider
from azure.identity.aio import AzureCliCredential
from azure.identity import AzureCliCredential
from azure.identity.aio import AzureCliCredential as AsyncAzureCliCredential
from redis.credentials import CredentialProvider
class AzureCredentialProvider(CredentialProvider):
"""Credential provider for Azure AD authentication with Redis Enterprise."""
def __init__(self, azure_credential: AzureCliCredential, user_object_id: str):
def __init__(self, azure_credential: AsyncAzureCliCredential, user_object_id: str):
self.azure_credential = azure_credential
self.user_object_id = user_object_id
@@ -57,24 +58,26 @@ async def main() -> None:
return
# Create Azure CLI credential provider (uses 'az login' credentials)
azure_credential = AzureCliCredential()
azure_credential = AsyncAzureCliCredential()
credential_provider = AzureCredentialProvider(azure_credential, user_object_id)
session_id = "azure_test_session"
# Create Azure Redis history provider
history_provider = RedisHistoryProvider(
source_id="redis_memory",
credential_provider=credential_provider,
host=redis_host,
port=10000,
ssl=True,
thread_id=session_id,
key_prefix="chat_messages",
max_messages=100,
)
# Create chat client
client = OpenAIChatClient()
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create agent with Azure Redis history provider
agent = client.as_agent(
python/samples/02-agents/context_providers/redis/redis_basics.py
+21 -6
View File
@@ -31,13 +31,16 @@ import asyncio
import os
from agent_framework import Message, tool
from agent_framework.openai import OpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.redis import RedisContextProvider
from azure.identity import AzureCliCredential
from redisvl.extensions.cache.embeddings import EmbeddingsCache
from redisvl.utils.vectorize import OpenAITextVectorizer
# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
# NOTE: approval_mode="never_require" is for sample brevity.
# Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py
# and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def search_flights(origin_airport_code: str, destination_airport_code: str, detailed: bool = False) -> str:
"""Simulated flight-search tool to demonstrate tool memory.
@@ -88,6 +91,15 @@ def search_flights(origin_airport_code: str, destination_airport_code: str, deta
)
def create_chat_client() -> AzureOpenAIResponsesClient:
"""Create an Azure OpenAI Responses client using a Foundry project endpoint."""
return AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
async def main() -> None:
"""Walk through provider-only, agent integration, and tool-memory scenarios.
@@ -100,8 +112,8 @@ async def main() -> None:
print("-" * 40)
# Create a provider with partition scope and OpenAI embeddings
# Please set the OPENAI_API_KEY and OPENAI_CHAT_MODEL_ID environment variables to use the OpenAI vectorizer
# Recommend default for OPENAI_CHAT_MODEL_ID is gpt-4o-mini
# Please set OPENAI_API_KEY to use the OpenAI vectorizer.
# For chat responses, also set AZURE_AI_PROJECT_ENDPOINT and AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME.
# We attach an embedding vectorizer so the provider can perform hybrid (text + vector)
# retrieval. If you prefer text-only retrieval, instantiate RedisContextProvider without the
@@ -115,6 +127,7 @@ async def main() -> None:
# scope data for multi-tenant separation; thread_id (set later) narrows to a
# specific conversation.
provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_basics",
application_id="matrix_of_kermits",
@@ -170,6 +183,7 @@ async def main() -> None:
)
# Recreate a clean index so the next scenario starts fresh
provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_basics_2",
prefix="context_2",
@@ -183,7 +197,7 @@ async def main() -> None:
)
# Create chat client for the agent
client = OpenAIChatClient(model_id=os.getenv("OPENAI_CHAT_MODEL_ID"), api_key=os.getenv("OPENAI_API_KEY"))
client = create_chat_client()
# Create agent wired to the Redis context provider. The provider automatically
# persists conversational details and surfaces relevant context on each turn.
agent = client.as_agent(
@@ -217,6 +231,7 @@ async def main() -> None:
print("-" * 40)
# Text-only provider (full-text search only). Omits vectorizer and related params.
provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_basics_3",
prefix="context_3",
@@ -227,7 +242,7 @@ async def main() -> None:
# Create agent exposing the flight search tool. Tool outputs are captured by the
# provider and become retrievable context for later turns.
client = OpenAIChatClient(model_id=os.getenv("OPENAI_CHAT_MODEL_ID"), api_key=os.getenv("OPENAI_API_KEY"))
client = create_chat_client()
agent = client.as_agent(
name="MemoryEnhancedAssistant",
instructions=(
python/samples/02-agents/context_providers/redis/redis_conversation.py
+8 -5
View File
@@ -17,8 +17,9 @@ Run:
import asyncio
import os
from agent_framework.openai import OpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.redis import RedisContextProvider
from azure.identity import AzureCliCredential
from redisvl.extensions.cache.embeddings import EmbeddingsCache
from redisvl.utils.vectorize import OpenAITextVectorizer
@@ -36,9 +37,8 @@ async def main() -> None:
cache=EmbeddingsCache(name="openai_embeddings_cache", redis_url="redis://localhost:6379"),
)
session_id = "test_session"
provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_conversation",
prefix="redis_conversation",
@@ -49,11 +49,14 @@ async def main() -> None:
vector_field_name="vector",
vector_algorithm="hnsw",
vector_distance_metric="cosine",
thread_id=session_id,
)
# Create chat client for the agent
client = OpenAIChatClient(model_id=os.getenv("OPENAI_CHAT_MODEL_ID"), api_key=os.getenv("OPENAI_API_KEY"))
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create agent wired to the Redis context provider. The provider automatically
# persists conversational details and surfaces relevant context on each turn.
agent = client.as_agent(
python/samples/02-agents/context_providers/redis/redis_sessions.py
+20 -19
View File
@@ -28,15 +28,24 @@ Run:
import asyncio
import os
import uuid
from agent_framework.openai import OpenAIChatClient
from agent_framework.azure import AzureOpenAIResponsesClient
from agent_framework.redis import RedisContextProvider
from azure.identity import AzureCliCredential
from redisvl.extensions.cache.embeddings import EmbeddingsCache
from redisvl.utils.vectorize import OpenAITextVectorizer
# Please set the OPENAI_API_KEY and OPENAI_CHAT_MODEL_ID environment variables to use the OpenAI vectorizer
# Recommend default for OPENAI_CHAT_MODEL_ID is gpt-4o-mini
# Please set OPENAI_API_KEY to use the OpenAI vectorizer.
# For chat responses, also set AZURE_AI_PROJECT_ENDPOINT and AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME.
def create_chat_client() -> AzureOpenAIResponsesClient:
"""Create an Azure OpenAI Responses client using a Foundry project endpoint."""
return AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
async def example_global_thread_scope() -> None:
@@ -44,20 +53,15 @@ async def example_global_thread_scope() -> None:
print("1. Global Thread Scope Example:")
print("-" * 40)
global_thread_id = str(uuid.uuid4())
client = OpenAIChatClient(
model_id=os.getenv("OPENAI_CHAT_MODEL_ID", "gpt-4o-mini"),
api_key=os.getenv("OPENAI_API_KEY"),
)
client = create_chat_client()
provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_threads_global",
application_id="threads_demo_app",
agent_id="threads_demo_agent",
user_id="threads_demo_user",
thread_id=global_thread_id,
scope_to_per_operation_thread_id=False, # Share memories across all sessions
)
@@ -97,10 +101,7 @@ async def example_per_operation_thread_scope() -> None:
print("2. Per-Operation Thread Scope Example:")
print("-" * 40)
client = OpenAIChatClient(
model_id=os.getenv("OPENAI_CHAT_MODEL_ID", "gpt-4o-mini"),
api_key=os.getenv("OPENAI_API_KEY"),
)
client = create_chat_client()
vectorizer = OpenAITextVectorizer(
model="text-embedding-ada-002",
@@ -109,6 +110,7 @@ async def example_per_operation_thread_scope() -> None:
)
provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_threads_dynamic",
# overwrite_redis_index=True,
@@ -165,10 +167,7 @@ async def example_multiple_agents() -> None:
print("3. Multiple Agents with Different Thread Configurations:")
print("-" * 40)
client = OpenAIChatClient(
model_id=os.getenv("OPENAI_CHAT_MODEL_ID", "gpt-4o-mini"),
api_key=os.getenv("OPENAI_API_KEY"),
)
client = create_chat_client()
vectorizer = OpenAITextVectorizer(
model="text-embedding-ada-002",
@@ -177,6 +176,7 @@ async def example_multiple_agents() -> None:
)
personal_provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_threads_agents",
application_id="threads_demo_app",
@@ -195,6 +195,7 @@ async def example_multiple_agents() -> None:
)
work_provider = RedisContextProvider(
source_id="redis_context",
redis_url="redis://localhost:6379",
index_name="redis_threads_agents",
application_id="threads_demo_app",
python/samples/02-agents/context_providers/simple_context_provider.py
+55 -64
View File
@@ -1,11 +1,13 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
from contextlib import suppress
from typing import Any
from agent_framework import Agent, AgentSession, BaseContextProvider, SessionContext, SupportsChatGetResponse
from agent_framework.azure import AzureAIClient
from azure.identity.aio import AzureCliCredential
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential
from pydantic import BaseModel
@@ -15,19 +17,13 @@ class UserInfo(BaseModel):
class UserInfoMemory(BaseContextProvider):
def __init__(self, client: SupportsChatGetResponse, user_info: UserInfo | None = None, **kwargs: Any):
def __init__(self, source_id: str = "user-info-memory", *, client: SupportsChatGetResponse, **kwargs: Any):
"""Create the memory.
If you pass in kwargs, they will be attempted to be used to create a UserInfo object.
"""
super().__init__("user-info-memory")
super().__init__(source_id)
self._chat_client = client
if user_info:
self.user_info = user_info
elif kwargs:
self.user_info = UserInfo.model_validate(kwargs)
else:
self.user_info = UserInfo()
async def after_run(
self,
@@ -38,12 +34,15 @@ class UserInfoMemory(BaseContextProvider):
state: dict[str, Any],
) -> None:
"""Extract user information from messages after each agent call."""
request_messages = context.get_messages()
# ensure you get all the messages you want to parse from, including the input in this case.
request_messages = context.get_messages(include_input=True, include_response=True)
# Check if we need to extract user info from user messages
user_messages = [msg for msg in request_messages if hasattr(msg, "role") and msg.role == "user"] # type: ignore
if (self.user_info.name is None or self.user_info.age is None) and user_messages:
try:
if (
state[self.source_id]["user_info"].name is None or state[self.source_id]["user_info"].age is None
) and user_messages:
with suppress(Exception):
# Use the chat client to extract structured information
result = await self._chat_client.get_response(
messages=request_messages, # type: ignore
@@ -53,17 +52,12 @@ class UserInfoMemory(BaseContextProvider):
)
# Update user info with extracted data
try:
with suppress(Exception):
extracted = result.value
if self.user_info.name is None and extracted.name:
self.user_info.name = extracted.name
if self.user_info.age is None and extracted.age:
self.user_info.age = extracted.age
except Exception:
pass # Failed to extract, continue without updating
except Exception:
pass # Failed to extract, continue without updating
if state[self.source_id]["user_info"].name is None and extracted.name:
state[self.source_id]["user_info"].name = extracted.name
if state[self.source_id]["user_info"].age is None and extracted.age:
state[self.source_id]["user_info"].age = extracted.age
async def before_run(
self,
@@ -74,55 +68,52 @@ class UserInfoMemory(BaseContextProvider):
state: dict[str, Any],
) -> None:
"""Provide user information context before each agent call."""
instructions: list[str] = []
if state.setdefault(self.source_id, None) is None:
state[self.source_id] = {"user_info": UserInfo()}
if self.user_info.name is None:
instructions.append(
"Ask the user for their name and politely decline to answer any questions until they provide it."
)
else:
instructions.append(f"The user's name is {self.user_info.name}.")
if self.user_info.age is None:
instructions.append(
"Ask the user for their age and politely decline to answer any questions until they provide it."
)
else:
instructions.append(f"The user's age is {self.user_info.age}.")
# Add context with additional instructions
context.extend_instructions(self.source_id, " ".join(instructions))
def serialize(self) -> str:
"""Serialize the user info for session persistence."""
return self.user_info.model_dump_json()
context.extend_instructions(
self.source_id,
"Ask the user for their name and politely decline to answer any questions until they provide it."
if state[self.source_id]["user_info"].name is None
else f"The user's name is {state[self.source_id]['user_info'].name}.",
)
context.extend_instructions(
self.source_id,
"Ask the user for their age and politely decline to answer any questions until they provide it."
if state[self.source_id]["user_info"].age is None
else f"The user's age is {state[self.source_id]['user_info'].age}.",
)
async def main():
async with AzureCliCredential() as credential:
client = AzureAIClient(credential=credential)
client = AzureOpenAIResponsesClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
deployment_name=os.environ["AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create the memory provider
memory_provider = UserInfoMemory(client)
context_name = "user-info-memory"
# Create the agent with memory
async with Agent(
client=client,
instructions="You are a friendly assistant. Always address the user by their name.",
context_providers=[memory_provider],
) as agent:
# Create a new session for the conversation
session = agent.create_session()
# Create the memory provider
memory_provider = UserInfoMemory(context_name, client=client)
print(await agent.run("Hello, what is the square root of 9?", session=session))
print(await agent.run("My name is Ruaidhrí", session=session))
print(await agent.run("I am 20 years old", session=session))
# Create the agent with memory
async with Agent(
client=client,
instructions="You are a friendly assistant. Always address the user by their name.",
context_providers=[memory_provider],
) as agent:
# Create a new session for the conversation
session = agent.create_session()
# Access the memory component and inspect the memories
if memory_provider:
print()
print(f"MEMORY - User Name: {memory_provider.user_info.name}")
print(f"MEMORY - User Age: {memory_provider.user_info.age}")
for msg in ["Hello, what is the square root of 9?", "My name is Ruaidhrí", "I am 20 years old"]:
print(f"User: {msg}")
print(f"Assistant: {await agent.run(msg, session=session)}")
# Access the memory component and inspect the memories
print()
print(f"MEMORY - User Name: {session.state[context_name]['user_info'].name}")
print(f"MEMORY - User Age: {session.state[context_name]['user_info'].age}")
if __name__ == "__main__":
python/samples/02-agents/providers/README.md
+19
View File
@@ -0,0 +1,19 @@
# Provider Samples Overview
This directory groups provider-specific samples for Agent Framework.
| Folder | What you will find |
| --- | --- |
| [`anthropic/`](anthropic/) | Anthropic Claude samples using both `AnthropicClient` and `ClaudeAgent`, including tools, MCP, sessions, and Foundry Anthropic integration. |
| [`amazon/`](amazon/) | AWS Bedrock samples using `BedrockChatClient`, including tool-enabled agent usage. |
| [`azure_ai/`](azure_ai/) | Azure AI Foundry V2 (`azure-ai-projects`) samples with `AzureAIClient`, from basic setup to advanced patterns like search, memory, A2A, MCP, and provider methods. |
| [`azure_ai_agent/`](azure_ai_agent/) | Azure AI Foundry V1 (`azure-ai-agents`) samples with `AzureAIAgentsProvider`, including provider methods and common hosted tool integrations. |
| [`azure_openai/`](azure_openai/) | Azure OpenAI samples for Assistants, Chat, and Responses clients, with examples for sessions, tools, MCP, file search, and code interpreter. |
| [`copilotstudio/`](copilotstudio/) | Microsoft Copilot Studio agent samples, including required environment/app registration setup and explicit authentication patterns. |
| [`custom/`](custom/) | Framework extensibility samples for building custom `BaseAgent` and `BaseChatClient` implementations, including layer-composition guidance. |
| [`foundry_local/`](foundry_local/) | Foundry Local samples using `FoundryLocalClient` for local model inference with streaming, non-streaming, and tool-calling patterns. |
| [`github_copilot/`](github_copilot/) | `GitHubCopilotAgent` samples showing basic usage, session handling, permission-scoped shell/file/url access, and MCP integration. |
| [`ollama/`](ollama/) | Local Ollama samples using `OllamaChatClient` (recommended) plus OpenAI-compatible Ollama setup, including reasoning and multimodal examples. |
| [`openai/`](openai/) | OpenAI provider samples for Assistants, Chat, and Responses clients, including tools, structured output, sessions, MCP, web search, and multimodal tasks. |
Each folder has its own README with setup requirements and file-by-file details.
python/samples/02-agents/providers/amazon/README.md
+17
View File
@@ -0,0 +1,17 @@
# Bedrock Examples
This folder contains examples demonstrating how to use AWS Bedrock models with the Agent Framework. The sample
uses `BEDROCK_CHAT_MODEL_ID`, `BEDROCK_REGION`, and AWS credentials (`AWS_ACCESS_KEY_ID`,
`AWS_SECRET_ACCESS_KEY`, optional `AWS_SESSION_TOKEN`).
## Examples
| File | Description |
|------|-------------|
| [`bedrock_chat_client.py`](bedrock_chat_client.py) | Uses `BedrockChatClient` with a simple tool-enabled `Agent` to demonstrate direct Bedrock chat integration. |
## Environment Variables
- `BEDROCK_CHAT_MODEL_ID`: Bedrock model ID (for example, `anthropic.claude-3-5-sonnet-20240620-v1:0`)
- `BEDROCK_REGION`: AWS region (defaults to `us-east-1` if unset)
- AWS credentials via standard variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, optional `AWS_SESSION_TOKEN`)
python/samples/02-agents/providers/amazon/bedrock_chat_client.py
+61
View File
@@ -0,0 +1,61 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from typing import Annotated
from agent_framework import Agent, tool
from agent_framework.amazon import BedrockChatClient
from pydantic import Field
"""
Bedrock Chat Client Example
This sample demonstrates using `BedrockChatClient` with an agent and a simple tool.
Environment variables used:
- `BEDROCK_CHAT_MODEL_ID`
- `BEDROCK_REGION` (defaults to `us-east-1` if unset)
- AWS credentials via standard variables (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`,
optional `AWS_SESSION_TOKEN`)
"""
# NOTE: approval_mode="never_require" is for sample brevity.
# Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py
# and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
city: Annotated[str, Field(description="The city to get the weather for.")],
) -> dict[str, str]:
"""Return a mock forecast for the requested city."""
normalized_city = city.strip() or "New York"
return {"city": normalized_city, "forecast": "72F and sunny"}
async def main() -> None:
"""Run a Bedrock-backed agent with one tool call."""
# 1. Create an agent with Bedrock chat client and one tool.
agent = Agent(
client=BedrockChatClient(),
instructions="You are a concise travel assistant.",
name="BedrockWeatherAgent",
tool_choice="auto",
tools=[get_weather],
)
# 2. Run a query that uses the weather tool.
query = "Use the weather tool to check the forecast for New York."
print(f"User: {query}")
response = await agent.run(query)
print(f"Assistant: {response.text}")
if __name__ == "__main__":
asyncio.run(main())
"""
Sample output:
User: Use the weather tool to check the forecast for New York.
Assistant: The forecast for New York is 72F and sunny.
"""
python/samples/02-agents/providers/anthropic/anthropic_claude_basic.py
+1 -1
View File
@@ -19,7 +19,7 @@ import asyncio
from typing import Annotated
from agent_framework import tool
from agent_framework_claude import ClaudeAgent
from agent_framework.anthropic import ClaudeAgent
@tool
python/samples/02-agents/providers/anthropic/anthropic_claude_with_mcp.py
+1 -1
View File
@@ -19,7 +19,7 @@ servers you trust. Use permission handlers to control what actions are allowed.
import asyncio
from typing import Any
from agent_framework_claude import ClaudeAgent
from agent_framework.anthropic import ClaudeAgent
from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny
python/samples/02-agents/providers/anthropic/anthropic_claude_with_multiple_permissions.py
+1 -1
View File
@@ -22,7 +22,7 @@ More permissions mean more potential for unintended actions.
import asyncio
from typing import Any
from agent_framework_claude import ClaudeAgent
from agent_framework.anthropic import ClaudeAgent
from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny
python/samples/02-agents/providers/anthropic/anthropic_claude_with_session.py
+1 -1
View File
@@ -13,7 +13,7 @@ from random import randint
from typing import Annotated
from agent_framework import tool
from agent_framework_claude import ClaudeAgent
from agent_framework.anthropic import ClaudeAgent
from pydantic import Field
python/samples/02-agents/providers/anthropic/anthropic_claude_with_shell.py
+1 -1
View File
@@ -14,7 +14,7 @@ Shell commands have full access to your system within the permissions of the run
import asyncio
from typing import Any
from agent_framework_claude import ClaudeAgent
from agent_framework.anthropic import ClaudeAgent
from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny
python/samples/02-agents/providers/anthropic/anthropic_claude_with_tools.py
+1 -1
View File
@@ -17,7 +17,7 @@ Available built-in tools:
import asyncio
from agent_framework_claude import ClaudeAgent
from agent_framework.anthropic import ClaudeAgent
async def main() -> None:
python/samples/02-agents/providers/anthropic/anthropic_claude_with_url.py
+1 -1
View File
@@ -16,7 +16,7 @@ URL fetching allows the agent to access any URL accessible from your network.
import asyncio
from agent_framework_claude import ClaudeAgent
from agent_framework.anthropic import ClaudeAgent
async def main() -> None:
python/samples/02-agents/providers/foundry_local/README.md
+22
View File
@@ -0,0 +1,22 @@
# Foundry Local Examples
This folder contains examples demonstrating how to run local models with `FoundryLocalClient` via `agent_framework.microsoft`.
## Prerequisites
1. Install Foundry Local and required local runtime components.
2. Install the connector package:
```bash
pip install agent-framework-foundry-local --pre
```
## Examples
| File | Description |
|------|-------------|
| [`foundry_local_agent.py`](foundry_local_agent.py) | Basic Foundry Local agent usage with streaming and non-streaming responses, plus function tool calling. |
## Environment Variables
- `FOUNDRY_LOCAL_MODEL_ID`: Optional model alias/ID to use by default when `model_id` is not passed to `FoundryLocalClient`.
python/samples/02-agents/providers/foundry_local/foundry_local_agent.py
+80
View File
@@ -0,0 +1,80 @@
# Copyright (c) Microsoft. All rights reserved.
# ruff: noqa
from __future__ import annotations
import asyncio
from random import randint
from typing import TYPE_CHECKING, Annotated
from agent_framework.microsoft import FoundryLocalClient
if TYPE_CHECKING:
from agent_framework import Agent
"""
This sample demonstrates basic usage of the FoundryLocalClient.
Shows both streaming and non-streaming responses with function tools.
Running this sample the first time will be slow, as the model needs to be
downloaded and initialized.
Also, not every model supports function calling, so be sure to check the
model capabilities in the Foundry catalog, or pick one from the list printed
when running this sample.
"""
def get_weather(
location: Annotated[str, "The location to get the weather for."],
) -> str:
"""Get the weather for a given location."""
conditions = ["sunny", "cloudy", "rainy", "stormy"]
return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."
async def non_streaming_example(agent: Agent) -> None:
"""Example of non-streaming response (get the complete result at once)."""
print("=== Non-streaming Response Example ===")
query = "What's the weather like in Seattle?"
print(f"User: {query}")
result = await agent.run(query)
print(f"Agent: {result}\n")
async def streaming_example(agent: Agent) -> None:
"""Example of streaming response (get results as they are generated)."""
print("=== Streaming Response Example ===")
query = "What's the weather like in Amsterdam?"
print(f"User: {query}")
print("Agent: ", end="", flush=True)
async for chunk in agent.run(query, stream=True):
if chunk.text:
print(chunk.text, end="", flush=True)
print("\n")
async def main() -> None:
print("=== Basic Foundry Local Client Agent Example ===")
client = FoundryLocalClient(model_id="phi-4-mini")
print(f"Client Model ID: {client.model_id}\n")
print("Other available models (tool calling supported only):")
for model in client.manager.list_catalog_models():
if model.supports_tool_calling:
print(
f"- {model.alias} for {model.task} - id={model.id} - {(model.file_size_mb / 1000):.2f} GB - {model.license}"
)
agent = client.as_agent(
name="LocalAgent",
instructions="You are a helpful agent.",
tools=get_weather,
)
await non_streaming_example(agent)
await streaming_example(agent)
if __name__ == "__main__":
asyncio.run(main())
python/samples/02-agents/providers/openai/openai_responses_client_image_generation.py
+11 -9
View File
@@ -6,7 +6,6 @@ import tempfile
import urllib.request as urllib_request
from pathlib import Path
import aiofiles # pyright: ignore[reportMissingModuleSource]
from agent_framework import Content
from agent_framework.openai import OpenAIResponsesClient
@@ -20,8 +19,11 @@ and automated visual asset generation.
"""
async def save_image(output: Content) -> None:
"""Save the generated image to a temporary directory."""
def save_image(output: Content) -> None:
"""Save the generated image to a temporary directory.
This sample is simplified, usually a async aware storing method would be better.
"""
filename = "generated_image.webp"
file_path = Path(tempfile.gettempdir()) / filename
@@ -37,15 +39,15 @@ async def save_image(output: Content) -> None:
data_bytes = None
else:
try:
data_bytes = await asyncio.to_thread(lambda: urllib_request.urlopen(uri).read())
data_bytes = urllib_request.urlopen(uri).read()
except Exception:
data_bytes = None
if data_bytes is None:
raise RuntimeError("Image output present but could not retrieve bytes.")
async with aiofiles.open(file_path, "wb") as f:
await f.write(data_bytes)
with open(file_path, "wb") as f:
f.write(data_bytes)
print(f"Image downloaded and saved to: {file_path}")
@@ -76,15 +78,15 @@ async def main() -> None:
image_saved = False
for message in result.messages:
for content in message.contents:
if content.type == "image_generation_tool_result_tool_result" and content.outputs:
if content.type == "image_generation_tool_result" and content.outputs:
output = content.outputs
if isinstance(output, Content) and output.uri:
await save_image(output)
save_image(output)
image_saved = True
elif isinstance(output, list):
for out in output:
if isinstance(out, Content) and out.uri:
await save_image(out)
save_image(out)
image_saved = True
break
if image_saved:
python/samples/02-agents/response_stream.py
+3 -5
View File
@@ -3,7 +3,7 @@
import asyncio
from collections.abc import AsyncIterable, Sequence
from agent_framework import ChatResponse, ChatResponseUpdate, Content, ResponseStream, Role
from agent_framework import ChatResponse, ChatResponseUpdate, Content, Message, ResponseStream
"""ResponseStream: A Deep Dive
@@ -256,8 +256,7 @@ async def main() -> None:
"""Result hook that wraps the response text in quotes."""
if response.text:
return ChatResponse(
messages=f'"{response.text}"',
role=Role.ASSISTANT,
messages=[Message(text=f'"{response.text}"', role="assistant")],
additional_properties=response.additional_properties,
)
return response
@@ -294,8 +293,7 @@ async def main() -> None:
# In real code, this would create an AgentResponse
text = "".join(u.text or "" for u in updates)
return ChatResponse(
text=f"[AGENT FINAL] {text}",
role=Role.ASSISTANT,
messages=[Message(text=f"[AGENT FINAL] {text}", role="assistant")],
additional_properties={"layer": "agent"},
)
python/samples/02-agents/typed_options.py
+7 -4
View File
@@ -22,6 +22,11 @@ which provides:
The sample shows usage with both OpenAI and Anthropic clients, demonstrating
how provider-specific options work for ChatClient and Agent. But the same approach works for other providers too.
The following environment variables are used:
- ANTHROPIC_API_KEY=...
- OPENAI_API_KEY=...
"""
@@ -109,14 +114,13 @@ async def demo_openai_chat_client_reasoning_models() -> None:
print("\n=== OpenAI ChatClient with TypedDict Options ===\n")
# Create OpenAI client
client = OpenAIChatClient[OpenAIReasoningChatOptions]()
client = OpenAIChatClient[OpenAIReasoningChatOptions](model_id="o3")
# With specific options, you get full IDE autocomplete!
# Try typing `client.get_response("Hello", options={` and see the suggestions
response = await client.get_response(
"What is 2 + 2?",
options={
"model_id": "o3",
"max_tokens": 100,
"allow_multiple_tool_calls": True,
# OpenAI-specific options work:
@@ -140,12 +144,11 @@ async def demo_openai_agent() -> None:
# or on the client when constructing the client instance:
# client = OpenAIChatClient[OpenAIReasoningChatOptions]()
agent = Agent[OpenAIReasoningChatOptions](
client=OpenAIChatClient(),
client=OpenAIChatClient(model_id="o3"),
name="weather-assistant",
instructions="You are a helpful assistant. Answer concisely.",
# Options can be set at construction time
default_options={
"model_id": "o3",
"max_tokens": 100,
"allow_multiple_tool_calls": True,
# OpenAI-specific options work: