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: Added AzureAI sample for downloading code interpreter generated files (#3189)

Browse Source 
* added azure ai code interpreter file download sample

* copilot fix suggestions

* function name fixes + readme update

* small fix
This commit is contained in:
Giles Odigwe
2026-01-14 14:50:07 -08:00
committed by GitHub
Unverified
parent 6c956ec596
commit 9b9a0f178c
3 changed files with 229 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/getting_started/agents/azure_ai/README.md
+1
View File
@@ -16,6 +16,7 @@ This folder contains examples demonstrating different ways to create and use age
| [`azure_ai_with_browser_automation.py`](azure_ai_with_browser_automation.py) | Shows how to use Browser Automation with Azure AI agents to perform automated web browsing tasks and provide responses based on web interactions. Requires a Browser Automation connection configured in your Azure AI project. |
| [`azure_ai_with_code_interpreter.py`](azure_ai_with_code_interpreter.py) | Shows how to use the `HostedCodeInterpreterTool` with Azure AI agents to write and execute Python code for mathematical problem solving and data analysis. |
| [`azure_ai_with_code_interpreter_file_generation.py`](azure_ai_with_code_interpreter_file_generation.py) | Shows how to retrieve file IDs from code interpreter generated files using both streaming and non-streaming approaches. |
| [`azure_ai_with_code_interpreter_file_download.py`](azure_ai_with_code_interpreter_file_download.py) | Shows how to download files generated by code interpreter using the OpenAI containers API. |
| [`azure_ai_with_existing_agent.py`](azure_ai_with_existing_agent.py) | Shows how to work with a pre-existing agent by providing the agent name and version to the Azure AI client. Demonstrates agent reuse patterns for production scenarios. |
| [`azure_ai_with_existing_conversation.py`](azure_ai_with_existing_conversation.py) | Demonstrates how to use an existing conversation created on the service side with Azure AI agents. Shows two approaches: specifying conversation ID at the client level and using AgentThread with an existing conversation ID. |
| [`azure_ai_with_application_endpoint.py`](azure_ai_with_application_endpoint.py) | Demonstrates calling the Azure AI application-scoped endpoint. |
python/samples/getting_started/agents/azure_ai/azure_ai_with_code_interpreter_file_download.py
+219
View File
@@ -0,0 +1,219 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
import tempfile
from pathlib import Path
from agent_framework import (
AgentResponseUpdate,
ChatAgent,
CitationAnnotation,
HostedCodeInterpreterTool,
HostedFileContent,
TextContent,
)
from agent_framework.azure import AzureAIProjectAgentProvider
from azure.identity.aio import AzureCliCredential
"""
Azure AI V2 Code Interpreter File Download Sample
This sample demonstrates how the AzureAIProjectAgentProvider handles file annotations
when code interpreter generates text files. It shows:
1. How to extract file IDs and container IDs from annotations
2. How to download container files using the OpenAI containers API
3. How to save downloaded files locally
Note: Code interpreter generates files in containers, which require both
file_id and container_id to download via client.containers.files.content.retrieve().
"""
QUERY = (
"Write a simple Python script that creates a text file called 'sample.txt' containing "
"'Hello from the code interpreter!' and save it to disk."
)
async def download_container_files(
file_contents: list[CitationAnnotation | HostedFileContent], agent: ChatAgent
) -> list[Path]:
"""Download container files using the OpenAI containers API.
Code interpreter generates files in containers, which require both file_id
and container_id to download. The container_id is stored in additional_properties.
This function works for both streaming (HostedFileContent) and non-streaming
(CitationAnnotation) responses.
Args:
file_contents: List of CitationAnnotation or HostedFileContent objects
containing file_id and container_id.
agent: The ChatAgent instance with access to the AzureAIClient.
Returns:
List of Path objects for successfully downloaded files.
"""
if not file_contents:
return []
# Create output directory in system temp folder
temp_dir = Path(tempfile.gettempdir())
output_dir = temp_dir / "agent_framework_downloads"
output_dir.mkdir(exist_ok=True)
print(f"\nDownloading {len(file_contents)} container file(s) to {output_dir.absolute()}...")
# Access the OpenAI client from AzureAIClient
openai_client = agent.chat_client.client
downloaded_files: list[Path] = []
for content in file_contents:
file_id = content.file_id
# Extract container_id from additional_properties
if not content.additional_properties or "container_id" not in content.additional_properties:
print(f" File {file_id}: ✗ Missing container_id")
continue
container_id = content.additional_properties["container_id"]
# Extract filename based on content type
if isinstance(content, CitationAnnotation):
filename = content.url or f"{file_id}.txt"
# Extract filename from sandbox URL if present (e.g., sandbox:/mnt/data/sample.txt)
if filename.startswith("sandbox:"):
filename = filename.split("/")[-1]
else: # HostedFileContent
filename = content.additional_properties.get("filename") or f"{file_id}.txt"
output_path = output_dir / filename
try:
# Download using containers API
print(f" Downloading {filename}...", end="", flush=True)
file_content = await openai_client.containers.files.content.retrieve(
file_id=file_id,
container_id=container_id,
)
# file_content is HttpxBinaryResponseContent, read it
content_bytes = file_content.read()
# Save to disk
output_path.write_bytes(content_bytes)
file_size = output_path.stat().st_size
print(f"({file_size} bytes)")
downloaded_files.append(output_path)
except Exception as e:
print(f"Failed: {e}")
return downloaded_files
async def non_streaming_example() -> None:
"""Example of downloading files from non-streaming response using CitationAnnotation."""
print("=== Non-Streaming Response Example ===")
async with (
AzureCliCredential() as credential,
AzureAIProjectAgentProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="V2CodeInterpreterFileAgent",
instructions="You are a helpful assistant that can write and execute Python code to create files.",
tools=HostedCodeInterpreterTool(),
)
print(f"User: {QUERY}\n")
result = await agent.run(QUERY)
print(f"Agent: {result.text}\n")
# Check for annotations in the response
annotations_found: list[CitationAnnotation] = []
# AgentResponse has messages property, which contains ChatMessage objects
for message in result.messages:
for content in message.contents:
if isinstance(content, TextContent) and content.annotations:
for annotation in content.annotations:
if isinstance(annotation, CitationAnnotation) and annotation.file_id:
annotations_found.append(annotation)
print(f"Found file annotation: file_id={annotation.file_id}")
if annotation.additional_properties and "container_id" in annotation.additional_properties:
print(f" container_id={annotation.additional_properties['container_id']}")
if annotations_found:
print(f"SUCCESS: Found {len(annotations_found)} file annotation(s)")
# Download the container files
downloaded_paths = await download_container_files(annotations_found, agent)
if downloaded_paths:
print("\nDownloaded files available at:")
for path in downloaded_paths:
print(f" - {path.absolute()}")
else:
print("WARNING: No file annotations found in non-streaming response")
async def streaming_example() -> None:
"""Example of downloading files from streaming response using HostedFileContent."""
print("\n=== Streaming Response Example ===")
async with (
AzureCliCredential() as credential,
AzureAIProjectAgentProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="V2CodeInterpreterFileAgentStreaming",
instructions="You are a helpful assistant that can write and execute Python code to create files.",
tools=HostedCodeInterpreterTool(),
)
print(f"User: {QUERY}\n")
file_contents_found: list[HostedFileContent] = []
text_chunks: list[str] = []
async for update in agent.run_stream(QUERY):
if isinstance(update, AgentResponseUpdate):
for content in update.contents:
if isinstance(content, TextContent):
if content.text:
text_chunks.append(content.text)
if content.annotations:
for annotation in content.annotations:
if isinstance(annotation, CitationAnnotation) and annotation.file_id:
print(f"Found streaming CitationAnnotation: file_id={annotation.file_id}")
elif isinstance(content, HostedFileContent):
file_contents_found.append(content)
print(f"Found streaming HostedFileContent: file_id={content.file_id}")
if content.additional_properties and "container_id" in content.additional_properties:
print(f" container_id={content.additional_properties['container_id']}")
print(f"\nAgent response: {''.join(text_chunks)[:200]}...")
if file_contents_found:
print(f"SUCCESS: Found {len(file_contents_found)} file reference(s) in streaming")
# Download the container files
downloaded_paths = await download_container_files(file_contents_found, agent)
if downloaded_paths:
print("\n✓ Downloaded files available at:")
for path in downloaded_paths:
print(f" - {path.absolute()}")
else:
print("WARNING: No file annotations found in streaming response")
async def main() -> None:
print("AzureAIClient Code Interpreter File Download Sample\n")
await non_streaming_example()
await streaming_example()
if __name__ == "__main__":
asyncio.run(main())
python/samples/getting_started/agents/azure_ai/azure_ai_with_code_interpreter_file_generation.py
+9 -9
View File
@@ -26,9 +26,9 @@ QUERY = (
)
async def test_non_streaming() -> None:
"""Test non-streaming response - should have annotations on TextContent."""
print("=== Testing Non-Streaming Response ===")
async def non_streaming_example() -> None:
"""Example of extracting file annotations from non-streaming response."""
print("=== Non-Streaming Response Example ===")
async with (
AzureCliCredential() as credential,
@@ -62,9 +62,9 @@ async def test_non_streaming() -> None:
print("WARNING: No file annotations found in non-streaming response")
async def test_streaming() -> None:
"""Test streaming response - check if file content is captured via HostedFileContent."""
print("\n=== Testing Streaming Response ===")
async def streaming_example() -> None:
"""Example of extracting file annotations from streaming response."""
print("\n=== Streaming Response Example ===")
async with (
AzureCliCredential() as credential,
@@ -106,9 +106,9 @@ async def test_streaming() -> None:
async def main() -> None:
print("AzureAIProjectAgentProvider Code Interpreter File Generation Test\n")
await test_non_streaming()
await test_streaming()
print("AzureAIClient Code Interpreter File Generation Sample\n")
await non_streaming_example()
await streaming_example()
if __name__ == "__main__":