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: update FoundryAgent for hosted agent sessions (#5447)

Browse Source 
* fixes to FoundryAgent to connect to new hosted agents

Co-authored-by: Copilot <copilot@github.com>

* fix mypy

Co-authored-by: Copilot <copilot@github.com>

* Python: remove Foundry service session helpers

Remove the public hosted-agent service session CRUD helpers from FoundryAgent and drop the related feature-stage inventory entry.

Update the hosted-agent sample to create and delete service sessions directly through the preview AIProjectClient APIs, and tighten a few test harnesses surfaced by full workspace validation.

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

* fix from merge

* fix hosted env detection

Co-authored-by: Copilot <copilot@github.com>

* reverted sample update

* fix tests and code

Co-authored-by: Copilot <copilot@github.com>

* remove aenter

* skipping some tests

Co-authored-by: Copilot <copilot@github.com>

---------

Co-authored-by: Copilot <copilot@github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard van Valkenburg
2026-04-24 11:25:03 +02:00
committed by GitHub
Unverified
parent 63c0a51797
commit 62e02da698
15 changed files with 601 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/samples/02-agents/chat_client/built_in_chat_clients.py
+1 -20
View File
@@ -75,11 +75,7 @@ def get_client(client_name: ClientName) -> SupportsChatGetResponse[Any]:
if client_name == "azure_openai_chat_completion":
return OpenAIChatCompletionClient(credential=AzureCliCredential())
if client_name == "foundry_chat":
return FoundryChatClient(
project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
model=os.environ["FOUNDRY_MODEL"],
credential=AzureCliCredential(),
)
return FoundryChatClient(credential=AzureCliCredential())
raise ValueError(f"Unsupported client name: {client_name}")
@@ -93,21 +89,6 @@ async def main(client_name: ClientName = "openai_chat") -> None:
print(f"Client: {client_name}")
print(f"User: {message.text}")
if isinstance(client, FoundryChatClient):
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
if stream:
response_stream = client.get_response([message], stream=True, options={"tools": get_weather})
print("Assistant: ", end="")
python/samples/04-hosting/foundry-hosted-agents/responses/README.md
+1 -1
View File
@@ -8,4 +8,4 @@ This folder contains a list of samples that show how to host agents using the `r
| [02_local_tools](./02_local_tools) | An example of hosting an agent with the `responses` API and local tools including a function tool and a local shell tool. |
| [03_remote_mcp](./03_remote_mcp) | An example of hosting an agent with the `responses` API and remote MCPs, including a GitHub MCP server and a Foundry Toolbox. |
| [04_workflows](./04_workflows) | An example of hosting a workflow with the `responses` API. |
| [using_deployed_agent.py](./using_deployed_agent.py) | An example of how to use the deployed agent in Agent Framework. |
| [using_deployed_agent.py](./using_deployed_agent.py) | Connect to the deployed basic Foundry agent with `FoundryAgent`, `allow_preview=True`, and version `v2`. |
python/samples/04-hosting/foundry-hosted-agents/responses/using_deployed_agent.py
+126 -30
View File
@@ -1,50 +1,146 @@
# Copyright (c) Microsoft. All rights reserved.
from __future__ import annotations
import asyncio
import os
from collections.abc import Mapping
from typing import Any, cast
from agent_framework import Agent, AgentResponse, AgentResponseUpdate, ResponseStream
from agent_framework.openai import OpenAIChatClient
from typing_extensions import Any
from agent_framework import AgentSession
from agent_framework.foundry import FoundryAgent
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import VersionRefIndicator
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
load_dotenv()
"""
This script demonstrates how to talk to a deployed agent using the OpenAIChatClient.
This sample demonstrates how to connect to the deployed basic Foundry agent with
`FoundryAgent`.
The sample uses environment variables for configuration, which can be set in a .env file or in the environment directly:
Environment variables:
FOUNDRY_PROJECT_ENDPOINT: Azure AI Foundry project endpoint.
FOUNDRY_AGENT_NAME: Hosted agent name.
FOUNDRY_AGENT_VERSION: Hosted agent version. Optional, defaults to latest if not specified.
After you deploy one of the agents in this directory, you can run this sample
to connect to it and have a conversation.
Note: The `allow_preview=True` flag is required to connect to the new hosted
agents, as this is a preview feature in Foundry.
Depending on where you have deployed your agent (local or Foundry Hosting), you may
need to change the base_url when initializing the OpenAIChatClient.
"""
async def print_streaming_response(streaming_response: ResponseStream[AgentResponseUpdate, AgentResponse[Any]]) -> None:
async for chunk in streaming_response:
if chunk.text:
print(chunk.text, end="", flush=True)
async def create_hosted_agent_session(
*,
agent: FoundryAgent,
project_client: AIProjectClient,
agent_name: str,
agent_version: str | None,
isolation_key: str,
) -> AgentSession:
"""Create a hosted-agent service session and wrap it in an AgentSession."""
create_session_kwargs: dict[str, Any] = {
"agent_name": agent_name,
"isolation_key": isolation_key,
}
resolved_agent_version = agent_version
if resolved_agent_version is None:
agent_details = await cast(Any, project_client.beta.agents).get( # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
agent_name=agent_name
)
versions = getattr(agent_details, "versions", None)
if not isinstance(versions, Mapping):
raise ValueError("Hosted agent details did not include a versions mapping.")
latest_version = getattr(cast(Any, versions.get("latest")), "version", None)
if not isinstance(latest_version, str) or not latest_version:
raise ValueError("Hosted agent details did not include a latest version string.")
resolved_agent_version = latest_version
create_session_kwargs["version_indicator"] = VersionRefIndicator(agent_version=resolved_agent_version)
service_session = await project_client.beta.agents.create_session(**create_session_kwargs)
agent_session_id = getattr(service_session, "agent_session_id", None)
if not isinstance(agent_session_id, str) or not agent_session_id:
raise ValueError("Hosted agent session creation did not return a non-empty agent_session_id.")
return agent.get_session(agent_session_id)
async def main() -> None:
agent = Agent(client=OpenAIChatClient(base_url="http://localhost:8088"))
session = agent.create_session()
credential = AzureCliCredential()
project_endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"]
agent_name = os.environ["FOUNDRY_AGENT_NAME"]
agent_version = os.getenv("FOUNDRY_AGENT_VERSION")
isolation_key = "my-isolation-key"
# First turn
query = "Hi!"
print(f"User: {query}")
print("Agent: ", end="", flush=True)
streaming_response = agent.run(query, session=session, stream=True)
await print_streaming_response(streaming_response)
project_client = AIProjectClient(
endpoint=project_endpoint,
credential=credential,
allow_preview=True,
)
async with (
project_client,
FoundryAgent(
project_client=project_client,
agent_name=agent_name,
agent_version=agent_version,
allow_preview=True,
) as agent,
):
session = await create_hosted_agent_session(
agent=agent,
project_client=project_client,
agent_name=agent_name,
agent_version=agent_version,
isolation_key=isolation_key,
)
# Second turn
query = "Your name is Javis. What can you do?"
print(f"\nUser: {query}")
print("Agent: ", end="", flush=True)
streaming_response = agent.run(query, session=session, stream=True)
await print_streaming_response(streaming_response)
try:
# 1. Send the first turn.
query = "Hi!"
print(f"User: {query}")
print("Agent: ", end="", flush=True)
async for chunk in agent.run(query, session=session, stream=True):
if chunk.text:
print(chunk.text, end="", flush=True)
# Third turn
query = "What is your name?"
print(f"\nUser: {query}")
print("Agent: ", end="", flush=True)
streaming_response = agent.run(query, session=session, stream=True)
await print_streaming_response(streaming_response)
# 2. Continue the conversation with the same deployed agent session.
query = "Your name is Javis. What can you do?"
print(f"\nUser: {query}")
print("Agent: ", end="", flush=True)
async for chunk in agent.run(query, session=session, stream=True):
if chunk.text:
print(chunk.text, end="", flush=True)
# 3. Ask a follow-up question in the same session.
query = "What is your name?"
print(f"\nUser: {query}")
print("Agent: ", end="", flush=True)
async for chunk in agent.run(query, session=session, stream=True):
if chunk.text:
print(chunk.text, end="", flush=True)
finally:
if session.service_session_id is not None:
await project_client.beta.agents.delete_session(
agent_name=agent_name,
session_id=session.service_session_id,
isolation_key=isolation_key,
)
if __name__ == "__main__":
asyncio.run(main())
"""
Sample output:
User: Hi!
Agent: Hello! How can I help you today?
User: Your name is Javis. What can you do?
Agent: I can answer questions and help with tasks using the instructions configured on the deployed agent.
User: What is your name?
Agent: My name is Javis.
"""