Foundry 托管智能体

Microsoft Foundry 代理服务中的托管代理允许将容器化代理应用程序部署到Microsoft托管的基础结构。 平台处理缩放、会话状态持久性、安全性和生命周期管理，以便你可以专注于代理的逻辑。 Microsoft Foundry 托管代理已正式发布，并支持使用自己的代码或首选代理框架生成的代理。 本文专门介绍代理框架托管集成。

通过使用 Agent Framework 托管集成，你可以用最少的代码通过 Foundry Responses 或 Invocations 协议公开 Agent。 Python还支持直接托管原生Workflow，而无需将其转换为代理。

注释

还可以使用 Azure 开发人员 CLI （azd） 工作流将与其他框架一起生成的代理代码部署到 Foundry 托管代理。 有关与框架无关的概念和部署指南，请参阅 什么是托管代理？ 本文的其余部分重点介绍 Agent Framework 集成。

何时使用托管代理

如果需要，请选择 Foundry 托管代理：

• 托管基础结构 - 无需自行配置容器、Web 服务器或缩放规则。
• 内置会话管理 — 平台在轮次和空闲期间保留 $HOME 和上传的文件。
• 专用代理标识 - 每个已部署的代理获取其自己的 Entra 标识，以便安全访问模型、工具和下游服务。
• 与 OpenAI 兼容的终结点 - 客户端可以通过响应协议使用任何与 OpenAI 兼容的 SDK 与代理交互。

相关场景

• 对于实时音频代理，请在 Foundry Tools（Voice Live）中使用结合 Azure 语音的托管代理，以实现服务器端语音活动检测、回声消除和降噪。 有关详细信息，请参阅 将 Voice Live 与托管代理配合使用。

注释

Python agent-framework-foundry-hosting 集成目前为预发布版本。 Microsoft Foundry 托管代理（托管托管服务）已正式发布。

Prerequisites

• Azure 订阅服务
• Azure Developer CLI (azd) 与 AI 智能体扩展：azd ext install azure.ai.agents

对于本地测试，还需要：

• 具有模型部署的 Microsoft Foundry 项目（例如，gpt-4o）
• 已安装并经过身份验证的 Azure CLI（az login）

• .NET 10 SDK 或更高版本

安装托管 NuGet 包：

dotnet add package Microsoft.Agents.AI.Foundry.Hosting --prerelease

• Python 3.10 或更高版本

安装预发行版托管包、Foundry 客户端和Azure身份验证包：

pip install --pre agent-framework-foundry agent-framework-foundry-hosting azure-identity

在 Foundry 中，平台提供调用方的用户上下文和调用上下文;托管基础结构使用它们来隔离每个用户的状态，并将请求上下文转发到 Foundry 服务。 本地运行时不会获得该平台的上下文信息，因此应用程序必须在需要时自行提供身份和状态控制机制。

响应协议

对于大多数代理，推荐以Responses协议为起点。 它公开与 OpenAI 兼容的 /responses 终结点，平台会自动管理会话历史记录、流式处理和会话生命周期。

对于Python托管代理，提前结束的响应具有incomplete状态。 流式处理客户端接收终端 response.incomplete 事件，而非流式处理客户端接收 status 设置为 incomplete。 content_filter 结束原因映射为设置为 content_filter 的 incomplete_details.reason，而 length 映射为 max_output_tokens。 任何生成的输出或拒绝内容仍可在响应中获取。

using Azure.AI.AgentServer.Core;
using Azure.AI.Projects;
using Azure.Identity;
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Foundry.Hosting;

var projectEndpoint = new Uri(Environment.GetEnvironmentVariable("FOUNDRY_PROJECT_ENDPOINT")
    ?? throw new InvalidOperationException("FOUNDRY_PROJECT_ENDPOINT is not set."));
var deployment = Environment.GetEnvironmentVariable("FOUNDRY_MODEL")
    ?? Environment.GetEnvironmentVariable("AZURE_AI_MODEL_DEPLOYMENT_NAME")
    ?? "gpt-4o";

AIAgent agent = new AIProjectClient(projectEndpoint, new DefaultAzureCredential())
    .AsAIAgent(
        model: deployment,
        instructions: "You are a helpful AI assistant.",
        name: "my-agent");

var builder = AgentHost.CreateBuilder(args);
builder.Services.AddFoundryResponses(agent);
builder.RegisterProtocol("responses", endpoints => endpoints.MapFoundryResponses());

var app = builder.Build();
app.Run();

AgentHost.CreateBuilder 创建为 Foundry 托管环境预配置的应用程序主机。 AddFoundryResponses 使用 Responses 协议处理程序注册你的智能体，并且 MapFoundryResponses 映射 /responses HTTP 终结点。

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import ResponsesHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a helpful AI assistant.",
)

server = ResponsesHostServer(agent)
server.run()

ResponsesHostServer 包装你的智能体并通过 Foundry Responses 协议公开它。 调用方的 store 字段控制是否保存外层响应以及由主机管理的会话和审批状态。 history_source 设置可单独选择由谁提供模型历史记录：

history_source	模型历史行为
"agent_server"（默认值）	主机重新构造存储的外部响应脚本，并禁用下游服务存储以防止重复历史记录。
"service"	主机仅发送当前输入，并在内部保存已启用存储的模型服务的延续 ID。 存储的提供程序会话无法从先前的响应进行分支。
"agent"	主机仅发送当前输入。 智能体的 HistoryProvider 或下游服务的默认存储设置负责管理历史记录。

请勿将 "agent_server" 或 "service" 与启用负载的 HistoryProvider 一起使用。 默认模式还会拒绝固定的下游延续选项，例如 conversation_id， previous_response_id和 conversation。 使用 history_source="agent" 实现自定义 SupportsAgentRun。

response_store构造函数参数为外部响应持久性选择后端。 较旧的构造函数参数 store 是 response_store 的已弃用别名；这两个参数都不会设置调用方的每请求 store 字段。 带有 store=false 的请求是一次性的：它不会保留主机管理的状态，会禁用受支持的下游存储，并且无法使用 background=true。

宿主拥有所提供的代理，并且可能会添加宿主特定的上下文提供程序。 不要将该代理复用于另一台主机，也不要在主机构造完成后直接调用它。

Responses 主机保留原生计算机调用、屏幕截图和安全检查功能。 应用程序必须执行请求的操作并显式确认任何安全检查。 有关完整流程，请参阅 原生计算机使用。

选择代理实例或工厂

ResponsesHostServer 和 InvocationsHostServer 都通过 agent 参数接受代理实例或无参数的同步或异步可调用对象。 主机在其整个生命周期内复用同一个实例。 可调用对象每个请求只运行一次，且返回的代理归属于该请求。

当常规代理在 AgentSession 之外保留可变状态时，请使用可调用项。 主机仅保存其支持的会话、检查点和函数审批存储，而不是请求范围的代理上的任意字段。

Warning

通过 agent= 托管 WorkflowAgent（例如 workflow.as_agent()）已弃用。 WorkflowAgent 会在两次运行之间将工作流状态保留在内存中，因此一个实例决不能为来自不同用户或对话的请求提供服务。 通过 workflow= 和可感知请求的工厂以原生方式托管工作流。

在迁移 Responses 主机之前，为每个请求创建一个新的工作流、执行程序、包装的代理、客户端、提供程序和工具的工厂是一种安全的传统形式。 使工作流名称和执行程序 ID 保持稳定，以便主机可以还原检查点。 对于 Invocations 主机，此旧工厂形式仅适用于无状态单轮工作流，因为代理托管不会还原工作流检查点。

当集成携带请求标识信息或管理请求专属资源时，也应使用工厂。 例如，如果这些对象使用当前的平台调用或用户上下文，则在工厂内部创建 MCP 连接、工具箱、技能提供程序、搜索客户端、内存提供程序及其凭据。 重用进程级的 MCP 连接可能会保留最初打开该连接的请求的身份信息。

主机针对每个请求进入和退出由工厂创建的智能体的上下文。 Agent 管理受上下文管理的客户端和 MCP 工具，但工厂函数必须负责关闭其创建的任何其他提供程序、传输机制或凭据。 不要关闭应用程序从工厂外部提供的共享对象。

使用 Responses 托管原生工作流

Python 可以通过 workflow= 直接托管已构建的工作流。 原生工作流需要一个 parse_response 回调，该回调将当前 Responses 请求映射为类型化的启动输入或完整的待处理回复批次：

from pydantic import BaseModel

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    HostedResponseRequest,
    ResponsesHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    text: str


def build_workflow(request: HostedResponseRequest):
    return build_fresh_workflow()


async def parse_response(request: HostedResponseRequest) -> WorkflowTurn[Ticket]:
    items = await request.get_input_items()
    if any(item.get("type") in ("function_call_output", "mcp_approval_response") for item in items):
        return WorkflowTurn(responses=await request.get_workflow_responses())

    text = await request.get_input_text()
    return WorkflowTurn(input=Ticket.model_validate_json(text or ""))


server = ResponsesHostServer(
    workflow=build_workflow,
    parse_response=parse_response,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[f"{Ticket.__module__}:{Ticket.__qualname__}"],
    ),
)

对可以暂停、继续或恢复后台工作的工作流使用请求感知同步或异步工厂。 工厂必须返回一个全新构建的图，其中包含全新的可变执行程序、代理、客户端、提供程序和工具。 使工作流名称和执行程序 ID 保持稳定，以便主机可以还原与外部响应关联的确切检查点。

受信任的平台用户和 Foundry 沙箱隔离原生工作流状态。 主机在使用任何回复权限之前验证完整的挂起答复批处理。 工作流执行之前，过时、部分、重复、重放、跨用户和跨沙盒答复会失败。 带有 store=false 的请求不会保存工作流状态，也无法返回可恢复的暂停

对于接受 list[Message] 的旧工作流，使用 response_input_messages(request) 来仅转换当前响应轮次。 它不会加载先前的外层历史记录或解码待处理的工作流回复。 托管 agent=workflow.as_agent() 在当前 beta 版期间仍可用，但会发出弃用警告。 有关完整示例，请参阅 原生 Responses 工作流示例。

持久保存状态并处理长时间运行的会话

ResponsesHostServer 并 InvocationsHostServer 默认配置持久性会话存储。 AgentSessionStoreProvider 提供一个 FoundryAgentSessionStore；响应会话使用 agent_sessions 逻辑存储，而调用会话使用单独的 invocation_sessions 存储。 这些存储区在托管环境中使用 Foundry 状态存储，在你本地运行时使用 SDK 的文件支持存储区。

对于 Responses 工作流代理，CheckpointStoreProvider 会提供一个 FoundryCheckpointStore。 原生 Responses 和 Invocations 工作流使用同一个提供程序来处理其精确的延续检查点。 FunctionApprovalStoreProvider 为待处理的智能体工具审批提供 FoundryFunctionApprovalStore。 原生工作流请求和审批回复则绑定到工作流检查点。

在 Foundry 中运行时，默认Python按平台用户 ID 和 Foundry 沙盒会话 ID 存储命名空间状态。 它们还要求每个状态操作都具有一个平台调用 ID。 调用 ID 授权并关联操作;它不是聊天 ID，不是存储密钥的一部分。

对于响应，平台配置的 FOUNDRY_AGENT_SESSION_ID 用于标识沙盒，而调用方提供的不同 agent_session_id 会被拒绝。 对于调用，宿主会根据请求上下文验证已路由的 agent_session_id 查询参数。 如果未配置 FOUNDRY_AGENT_SESSION_ID，则查询参数必须存在、非空，并且与请求上下文匹配。 缺失、重复或冲突的值会被拒绝，而不是使用 SDK 回退 ID。

这些保证适用于默认托管存储区。 自定义存储提供程序必须实现等效的用户和沙盒隔离，保留内部 AgentSession.session_id 与主机查找密钥分开，并使用条件写入，以便过时的请求无法覆盖较新的快照。 新密钥应使用仅创建写入，而不是无条件的向上插入。 请参阅包含 ETag 保护的写入和删除的 Cosmos DB 实现的 自定义存储示例 。

使用 history_source="agent" 时，配置的会话存储会持久化由 AgentSession 携带的提供程序状态，包括来自 InMemoryHistoryProvider 的消息。

两个主机都通过 agent_session_store_provider 接受 StoreProvider[SessionStore]。 会话状态必须支持 AgentSession 序列化。 为自定义状态类型register_state_type()注册编解码器;还原状态不会保留Python对象标识。 新的默认存储中的会话将在上次写入 30 天后过期。 自定义提供程序控制其自己的保留期。

有作用域的默认存储不会读取旧版无作用域的 agent_sessions、invocation_sessions、检查点或函数批准数据。 开始一个全新的 Responses 对话，而不要复用旧的 previous_response_id 或 conversation ID。 Invocations 在作用域存储中以一个空的 Agent Framework 会话开始。

加载的 AgentSession 记录使用 ETag 条件。 如果另一个请求先推进了同一会话，过时的写入就会失败，而不会覆盖更新的状态。 此检查不会为代理或工具产生的副作用提供事务保障或恰好一次执行保障，因此应用程序仍必须协调重叠的请求。

对于 Responses 专用存储，请将 StoreProvider 传递给 function_approval_store_provider，或将 ContextScopedStoreProvider 传递给 checkpoint_store_provider。

外部后台任务使用调用方可见的 response.id 进行轮询。 默认情况下，background_source="agent_server" 会使后台执行保留在主机端。 仅在使用 history_source="service" 以及已启用存储且可恢复的 Responses 客户端时设置 background_source="provider"。 如果 ResponsesServerOptions(resilient_background=True) 也已设置，则主机只有在保存私有延续令牌后才能恢复提供程序轮询。 确保本地工具的副作用具有幂等性，因为如果在下一个词元保存之前发生崩溃，这些副作用可能会被再次执行。

从azure.ai.agentserver.responses中导入ResponsesServerOptions，并将其传递给ResponsesHostServeroptions参数。 可用的长时间运行的会话选项取决于代理类型：

能力	代理类型	要求和行为
工作流检查点后台恢复	仅限工作流	设置 ResponsesServerOptions(resilient_background=True)。 使用 background=true 和 store=true 发送 Responses 请求。 重启后，主机将恢复最新的持久工作流检查点，或者在不存在检查点时重播原始输入。 不要在工作流上配置检查点存储，因为主机管理它。 确保外部副作用具有幂等性，因为最后一个持久检查点之后的处理可能会重复执行。
提供商原生后台响应	使用已启用存储的 Responses 客户端的非工作流 Agent	设置 history_source="service" 和 background_source="provider"。 如果保存的提供程序延续令牌必须在主机重启后仍然保留，请设置 resilient_background=True。
可控对话	暂时不可用	不要设置 steerable_conversations=True。 在 Agent Server SDK 能够安全处理被拒绝的引导轮次之前，主机会在构造期间引发 RuntimeError。

有关完整的实现，请参阅 自定义存储、 基本响应历史记录和背景，以及 可复原的长时间运行的工作流 示例。

从托管沙盒读取文件

将托管沙盒的持久性 $HOME 视为请求路由资源，而不是常规文件系统边界。 仅接受由您的应用程序显式上传到专用目录的文件，验证当前沙盒身份，并拒绝绝对路径、路径遍历、链接、非常规文件以及过大或无效的内容。

对于 Responses 协议，请使用 agent_session_id 正文字段将请求路由到托管会话。 查询字符串选择器用于 Invocations。 会话上传和工具箱代码解释器文件是单独的资源;上传的沙盒文件不会自动装载到工具箱容器中。 有关绑定的 UTF-8 读取和本地和托管上传指南，请参阅 会话文件示例 。

控制请求选项

主机将本机响应生成字段映射到 Agent Framework 运行选项。 例如，max_output_tokens 变为 max_tokens，parallel_tool_calls 变为 allow_multiple_tool_calls。 来自 extra_body 的扁平化值会覆盖已翻译的原生值。

在常规代理运行之前，使用同步或异步 prepare_options(request, options) 挂钩删除或替换调用方模型选项。 钩子无法设置由主机控制的身份、存储、延续或传输字段。 对于不接受运行时模型选项的自定义 SupportsAgentRun 实现，请设置为 unsupported_options"warn" （默认值）， "ignore"或 "error"。

处理 OAuth 同意请求

当 Foundry 托管的 MCP 工具需要用户同意时， ResponsesHostServer 返回包含 oauth_consent_request 输出项的不完整响应。 向用户展示其 consent_link，然后在用户完成同意操作后，以 previous_response_id 作为该不完整响应的 ID 继续处理。 主机端会在此次重试期间保留代理会话，并且仅提供绝对 HTTPS 同意链接。

如果主机知道预期的授权来源，请使用 allowed_oauth_consent_origins 限制同意链接：

server = ResponsesHostServer(
    agent,
    allowed_oauth_consent_origins=[
        "https://logic-region.consent.azure-apihub.net",
        "https://auth.partner.example",
    ],
)

省略允许列表会保留绝对 HTTPS 验证，而不会限制目标源。 提供空列表会拒绝每个同意链接。 仅配置确切的 HTTPS 源;包含路径、查询或片段的条目将被拒绝。

调用协议

调用协议可让你完全控制 HTTP 请求和响应。 如果您需要自定义负载、非会话处理或不兼容 OpenAI 的流式处理协议，请使用它。

使用 C# 中的调用协议，可以实现用于处理传入请求的自定义 InvocationHandler ：

using Azure.AI.AgentServer.Core;
using Azure.AI.AgentServer.Invocations;
using Microsoft.Agents.AI;

var builder = AgentHost.CreateBuilder(args);

builder.Services.AddSingleton<AIAgent, MyAgent>();
builder.Services.AddInvocationsServer();
builder.Services.AddScoped<InvocationHandler, MyInvocationHandler>();

builder.RegisterProtocol("invocations", endpoints => endpoints.MapInvocationsServer());

var app = builder.Build();
app.Run();

该方法 AddInvocationsServer 注册调用协议服务。 实现 InvocationHandler 以定义代理处理每个请求的方式。

对于轻量级设置，请使用 agent_framework_foundry_hosting 包中的 InvocationsHostServer。 它像 ResponsesHostServer 一样封装您的代理，并自动处理会话管理。

import os

from agent_framework import Agent
from agent_framework.foundry import FoundryChatClient
from agent_framework_foundry_hosting import InvocationsHostServer
from azure.identity import DefaultAzureCredential

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

server = InvocationsHostServer(agent)
server.run()

InvocationsHostServer 接受为 Responses 主机所描述的相同的实例或请求范围工厂形式。 它从配置的存储中还原序列化会话，因此完成的对话可以在主机重启后继续。 有关存储行为、保留和自定义，请参阅 “持久”状态并处理长时间运行的会话。

托管时，调用使用持久状态中所述的已验证请求范围 并处理长时间运行的会话。 视为 AgentSession.session_id 一个不透明值;不分析或依赖于其内部表示形式。 本地运行保留其现有的单用户存储行为。

使用 Invocations 托管原生工作流

传递 workflow= 和显式的 parse_request 回调以托管原生工作流。 回调拥有应用程序 JSON 架构，并返回一个包含类型化输入或完整待回复批次的 WorkflowTurn：

from pydantic import BaseModel
from starlette.requests import Request

from agent_framework_foundry_hosting import (
    CheckpointStoreProvider,
    InvocationsHostServer,
    WorkflowTurn,
)


class Ticket(BaseModel):
    ticket_id: str
    question: str


class TicketDecision(BaseModel):
    approved: bool


def build_workflow(_request: Request):
    return build_fresh_workflow()


async def parse_request(request: Request) -> WorkflowTurn[Ticket]:
    payload = await request.json()
    stream = payload.get("stream", False)

    if "responses" in payload:
        decisions = {
            request_id: TicketDecision.model_validate(value)
            for request_id, value in payload["responses"].items()
        }
        return WorkflowTurn(responses=decisions, stream=stream)

    ticket = Ticket.model_validate(payload)
    return WorkflowTurn(input=ticket, stream=stream)


server = InvocationsHostServer(
    workflow=build_workflow,
    parse_request=parse_request,
    checkpoint_store_provider=CheckpointStoreProvider(
        allowed_checkpoint_types=[
            f"{Ticket.__module__}:{Ticket.__qualname__}",
            f"{TicketDecision.__module__}:{TicketDecision.__qualname__}",
        ],
    ),
)

包括工作流保存在检查点提供程序 allowed_checkpoint_types 列表中的每个自定义应用程序类型。

托管工作流需要一个请求感知工厂，该工厂返回具有稳定工作流和执行器 ID 的新构建图。 直接构建的工作流仅适用于不会暂停的本地单次运行。

非流式工作流响应采用 JSON 格式，并包含一个 output 事件列表。 流式处理会发出带帧的 output 和 request_info 事件，只有在精确的工作流游标保存后，才会随后发出 done。 将流式输出视为临时输出，直到 done。 原生工作流不支持 legacy_wire_format=True。

主机根据受信任的用户和沙盒范围内的确切挂起检查点验证答复。 如果工作流有多个挂起的请求，请在一次轮次中回复完整的批处理。 有关可运行的解析器、类型化工单工作流、检查点类型允许列表和 JSON/SSE 示例，请参阅 原生 Invocations 工作流示例。

自定义调用请求和响应

默认情况下， POST /invocations 接受包含字符串 message、可选对象和可选 options 布尔 stream 值的 JSON 对象。 若要接受特定于应用程序的有效负载，请传递一个返回 parse_request 的同步或异步 InvocationRun(messages, options, stream) 回调。 使用 prepare_options 可在代理运行前筛选或替换调用方生成选项的副本。

主机验证挂钩输出，并拒绝平台标识、存储、延续和代理执行控制。 对于不接受运行时选项的代理，设置为 unsupported_options"warn" （默认值）， "ignore"或 "error"。 有关完整实现，请参阅 调用分析器示例 。

非流式请求成功时，返回格式为 {"response": "..."} 的 JSON。 流式处理使用服务器发送事件：一个或多个 event: delta 帧，随后在成功时跟随 event: done，在失败时跟随 event: error。 流可以在错误发生前发出增量内容，因此，客户端必须将 done（而不是增量内容）视为成功完成的标志。 主机仅在完成响应流并持久保存done后，才发出AgentSession。 它的 session_id 是平台沙盒路由 ID，而不是序列化后的 AgentSession.session_id。

仅在迁移需要先前纯文本响应和原始文本块流的现有客户端时，才将 legacy_wire_format=True 设为该值。 此兼容性模式已弃用，不会将失败转换为成功的文本。 主机仅在单个进程内对同一会话中的请求进行串行化处理；在外部工具产生影响后，仍可能发生跨进程的比较并交换冲突。

调用协议不会继续执行处于挂起状态或已中断的工作流运行。 需要不同的工作流延续行为时，请使用以下部分中的自定义处理程序模式。

若要完全控制请求处理，请直接从InvocationAgentServerHost包使用azure.ai.agentserver.invocations并实现自己的调用处理程序：

import os
from collections.abc import AsyncGenerator

from agent_framework import Agent, AgentSession
from agent_framework.foundry import FoundryChatClient
from azure.ai.agentserver.invocations import InvocationAgentServerHost
from azure.identity import DefaultAzureCredential
from starlette.requests import Request
from starlette.responses import JSONResponse, Response, StreamingResponse

_sessions: dict[str, AgentSession] = {}

client = FoundryChatClient(
    project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"],
    model=os.environ.get("FOUNDRY_MODEL") or os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
    credential=DefaultAzureCredential(),
)

agent = Agent(
    client=client,
    instructions="You are a friendly assistant. Keep your answers brief.",
    default_options={"store": False},
)

app = InvocationAgentServerHost()


@app.invoke_handler
async def handle_invoke(request: Request):
    """Handle streaming multi-turn chat."""
    data = await request.json()
    session_id = request.state.session_id
    stream = data.get("stream", False)
    user_message = data.get("message", None)

    if user_message is None:
        return Response(content="Missing 'message' in request", status_code=400)

    session = _sessions.setdefault(session_id, AgentSession(session_id=session_id))

    if stream:

        async def stream_response() -> AsyncGenerator[str]:
            async for update in agent.run(user_message, session=session, stream=True):
                yield update.text

        return StreamingResponse(
            stream_response(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
        )

    response = await agent.run([user_message], session=session, stream=stream)
    return JSONResponse({"response": response.text})


if __name__ == "__main__":
    app.run()

Warning

自定义处理程序示例中的内存中会话存储在重启时丢失。 在生产中使用持久存储（例如 Cosmos DB）。

如需了解完整的 Invocations 部署，请参阅 Foundry 托管的 Telegram 示例。 它将 API 管理置于托管代理 Webhook 的前面，并使用托管标识、密钥保管库和 Cosmos DB 进行持久聊天历史记录。

注释

即将推出对 Foundry 托管代理的 Go 支持。 有关最新状态，请参阅 Agent Framework Go 存储库 。

Tip

有关托管代理项目的示例，请参阅Python示例或 C# 示例。 azd ai agent init或使用该命令从头开始搭建新的托管代理项目。 有关分步说明，请参阅本 快速入门指南 。

在本地运行

Azure开发人员 CLI（azd）提供了在本地运行和测试托管代理的最简单方法。

初始化项目

创建新文件夹，并从示例的 azure.yaml 文件初始化它。 在本演练中使用容器部署：

mkdir my-hosted-agent && cd my-hosted-agent
azd ai agent init -t <path-to-azure.yaml> --src . --deploy-mode container

Tip

替换<path-to-azure.yaml>示例的azure.yaml项目文档为本地路径或受支持的 GitHub URL。

设置环境变量。

export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export FOUNDRY_MODEL="<your-model-deployment>"

运行代理主机

azd ai agent run

代理主机在http://localhost:8088启动。

调用代理

azd ai agent invoke --local "Hello!"

或者使用curl：

curl -X POST http://localhost:8088/responses \
  -H "Content-Type: application/json" \
  -d '{"input": "Hello!"}'

或在 PowerShell 中：

(Invoke-WebRequest -Uri http://localhost:8088/responses -Method POST -ContentType "application/json" -Body '{"input": "Hello!"}').Content

部署到 Foundry

在本地验证代理后，将其部署到 Microsoft Foundry：

1. 预配资源（如果您还没有 Foundry 项目）：

   azd provision
   

   这会创建一个资源组，其中包含 Foundry 实例、项目、模型部署、Application Insights 和容器注册表。

2. 部署代理：

   azd deploy
   

   这会将代理打包为容器映像，将其推送到Azure 容器注册表，并将其部署到 Foundry 代理服务。

Foundry 托管基础结构会在运行时自动将以下环境变量注入代理容器：

Variable	说明
FOUNDRY_PROJECT_ENDPOINT	Foundry 项目的端点 URL。
AZURE_AI_MODEL_DEPLOYMENT_NAME	在 azd ai agent init 期间配置的 azd 管理的模型部署名称。 Python代码可以优先在本地使用FOUNDRY_MODEL，并回退到此托管值。
APPLICATIONINSIGHTS_CONNECTION_STRING	用于遥测的 Application Insights 连接字符串。

部署后，代理可通过其专用 Foundry 终结点进行访问，也可以从 Foundry 门户进行测试。

后续步骤

托管代理概念

• 使用 Foundry SDK 部署托管代理
• 管理托管代理
• 耐用扩展
• 自托管 A2A 代理
• Python 示例
• C# 示例