Foundry 託管代理程式

• .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

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("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["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 設定可獨立選擇由誰提供模型歷程：

請勿將 "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 主機保留原生電腦通話、截圖及安全檢查。 您的應用程式必須執行所要求的動作，並明確確認任何安全檢查。 完整流程請參見 原生電腦使用。

選擇代理實例或工廠

InvocationsHostServer 和 ResponsesHostServer 都可透過 agent 參數接受代理程式執行個體，或是不接受任何引數的同步或非同步可呼叫物件。 主機會在其整個生命週期內重複使用同一個實例。 可呼叫物件每個請求只會執行一次，且回傳的代理屬於該請求。

當代理程式在 AgentSession 外部保留可變狀態時，請使用可呼叫物件。 特別是，從可建立全新工作流程、執行器和封裝代理程式的工廠建立一個 WorkflowAgent：

def create_workflow_agent():
    return build_workflow().as_agent(name="support-workflow")


server = ResponsesHostServer(agent=create_workflow_agent)

保持工作流程名稱和執行器 ID 穩定，讓後續回應請求能找到已儲存的檢查點。 ResponsesHostServer 透過其工作階段、檢查點和函式核准存放區維持受支援的狀態；它不會將任意欄位持續儲存在以要求為範圍的 Agent 上。 請參閱 工作流程 與 韌性長期執行工作流程 範例。

當整合會攜帶要求識別資訊，或持有要求專屬的資源時，也要使用工廠。 例如，在工廠內建立 MCP 連線、工具箱、技能提供者、搜尋客戶端、記憶體提供者，以及它們在使用當前平台呼叫或使用者上下文時的憑證。 重複使用整個程序共用的 MCP 連線，可能會保留開啟該連線之請求的身分。

針對每個請求，主機都會進入並離開由工廠建立的代理程式。 Agent 會管理受情境管理的用戶端和 MCP 工具，但你的工廠必須關閉或清理它所建立的任何其他提供者、傳輸機制或憑證。 不要關閉應用程式從工廠外提供的共用物件。

用 Responses 來執行原生工作流程

Python 可以直接透過 workflow= 裝載已建構的工作流程。 原生工作流程需要一個 parse_response 回呼，將目前的回應請求映射到具型別的起始輸入或完整的待處理回覆批次：

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)，僅轉換目前的 Responses 回合。 它不會載入之前的外部歷程，也不會解碼待處理的工作流程回覆。 託管 agent=workflow.as_agent() 在目前測試期間仍可使用，但會發出棄用警告。 完整範例請參閱 原生 Responses 工作流程範例。

持續狀態並處理長時間對話

ResponsesHostServer 和 InvocationsHostServer 預設會設定持久性工作階段儲存體。 AgentSessionStoreProvider 提供 FoundryAgentSessionStore；Responses 工作階段使用 agent_sessions 邏輯存放區，而 Invocations 工作階段使用獨立的 invocation_sessions 存放區。 這些存放區在託管時會使用 Foundry State Store，而在本機執行時則會使用 SDK 的檔案型儲存體。

對於 Responses 工作流程代理程式，CheckpointStoreProvider 會提供一個 FoundryCheckpointStore。 原生回應與調用工作流程使用相同的提供者來執行其對應的續接檢查點。 FunctionApprovalStoreProvider 提供待審核代理工具核准的 FoundryFunctionApprovalStore。 原生工作流程請求與核准回覆則綁定在工作流程檢查點。

在 Foundry 中執行時，預設的 Python 會以平台使用者 ID 和 Foundry 沙箱會話 ID 儲存命名空間狀態。 他們也要求每個州的營運都設定一個平台呼叫ID。 通話識別碼授權並關聯操作;它不是對話 ID，也不是儲存金鑰的一部分。

對於回應，平台配置 FOUNDRY_AGENT_SESSION_ID 會識別沙盒，並拒絕另一個呼叫者提供的 agent_session_id 呼叫。 對於叫用，主機會依據要求上下文驗證經路由處理的agent_session_id查詢參數 。 如果 FOUNDRY_AGENT_SESSION_ID 未設定，查詢參數必須存在、非空且符合請求上下文。 缺少、重複或衝突的值會被拒絕，而不是使用 SDK 備援 ID。

這些保證適用於預設的託管商店。 自訂儲存提供者必須實現等效的使用者與沙盒隔離，將內部 AgentSession.session_id 資料與主機查詢鍵分開保存，並使用條件寫入，避免過時的請求覆蓋較新的快照。 新鍵應該只用創建寫入，而非無條件的上溢點。 請參閱 Cosmos DB 中具備 ETag 保護寫入與刪除功能的 自訂儲存範例 。

使用 history_source="agent" 時，已設定的工作階段儲存區會持久保存由 AgentSession 攜帶的提供者狀態，包括來自 InMemoryHistoryProvider 的訊息。

兩個主機都接受一個從 StoreProvider[SessionStore] 到 agent_session_store_provider 的範圍。 會話狀態必須支援序列 AgentSession 化。 使用 register_state_type() 為自訂狀態類型註冊編解碼器；還原的狀態不會保留 Python 物件識別。 新的預設儲存會在最後一次寫入後 30 天內結束會話。 自訂提供者自行控管其保留設定。

具作用域的預設存放區不會讀取舊的無範圍agent_sessionsinvocation_sessions、檢查點或函式核准資料。 請開始新的 Responses 對話，而不要重複使用舊的 previous_response_id 或對話 ID。 叫用從作用域存放區中的空 Agent Framework 工作階段開始

已載入的AgentSession紀錄會套用 ETag 條件。 如果另一個請求先讓同一工作階段往前推進，過時的寫入就會失敗，而不會覆寫較新的狀態。 此檢查不提供代理或工具副作用的交易或精確一次執行，因此應用程式仍需協調重疊請求。

對於回應專用的儲存，請將 a StoreProvider 傳遞到 function_approval_store_provider 或 a ContextScopedStoreProvider 傳遞給 checkpoint_store_provider。

外部背景作業使用對呼叫端可見的 response.id 來進行輪詢。 預設 background_source="agent_server" 會保留主機的背景執行。 僅能在將background_source="provider"、history_source="service"以及可存放、可繼續的 Responses 用戶端搭配時才能設定。 若 ResponsesServerOptions(resilient_background=True) 也設定為 ，主機只能在儲存私有延續標記後恢復提供者輪詢。 讓本地工具的副作用具備冪等性，因為如果在儲存下一個 token 之前發生當機，就可能再次執行這些副作用。

從 azure.ai.agentserver.responses 匯入 ResponsesServerOptions，並透過 ResponsesHostServer 參數將其傳遞給 options。 可用的長時間對話選項依代理類型而異：

完整實作請參閱 自訂儲存、 基本回應歷史與背景，以及 韌性長期執行的工作流程 範例。

從託管沙盒讀取檔案

將託管沙盒的持久化 $HOME 視為請求路由資源，而非一般的檔案系統邊界。 只接受應用程式明確上傳到專用目錄的檔案，驗證目前沙盒身份，並拒絕絕對路徑、遍歷、連結、非規則檔案以及過大或無效內容。

對於回應協定，將請求路由到帶有 agent_session_id 主體欄位的託管會話。 查詢字串選擇器供叫用使用。 會話上傳與 Toolbox 程式碼直譯器檔案是分開的資源;上傳的沙盒檔案不會自動掛載到 Toolbox 容器中。 請參閱 會話檔案範例 ，了解有界 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 。 主機會保留代理會話以供此重試使用，並只公開絕對的 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 來源;包含路徑、查詢或片段的條目會被拒絕。

使用 C# 中的 Invocations 協定，您可以實現一個自訂 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 來定義代理如何處理每個請求。

想要輕量化的配置，請使用 InvocationsHostServer 套件中的 agent_framework_foundry_hosting。 它會將您的代理類似於 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["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 接受與回應主機描述相同的實例或請求範圍的工廠表單。 它會從已設定的儲存庫還原序列化的會話，讓完成的對話在主機重啟後仍能繼續。 關於儲存行為、保留與自訂，請參見 持久狀態與處理長期對話。

當被託管時，Invocations 會使用持久狀態中描述的已驗證請求範圍 ，並處理長時間執行的對話。 將 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。 直接建構的工作流程僅適用於本地且不暫停的單次執行。

非串流工作流程回應使用帶有 output 事件清單的 application/json。 串流會輸出 frameed 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"。 完整實作請參見 Invocations 解析器範例 。

非串流成功時會以如下的 JSON 格式傳回：{"response": "..."} 串流使用伺服器傳送事件：先傳送一個或多個 event: delta 框格，接著在成功時傳送 event: done，或在失敗時傳送 event: error。 串流在發生錯誤前可能會先傳送 delta，因此用戶端必須將 done 而非 delta 視為成功完成。 主機只有在完成回應串流並持續執行AgentSession後才會發出done。 它的 session_id 是平台沙盒路由 ID，不是序列化的 AgentSession.session_id。

僅在遷移需要先前純文字回應與原始文字區塊串流的現有用戶端時設定 legacy_wire_format=True 。 此相容模式已被棄用，且無法將失敗轉換為成功的文字。 主機僅會在單一行程內將同一工作階段的請求序列化；但在外部工具造成影響之後，仍可能發生跨行程的比較並交換衝突。

叫用通訊協定不會繼續執行處於待處理或已中斷狀態的工作流程執行。 當你需要不同的工作流程延續行為時，請使用以下章節的自訂處理模式。

為了完全掌控要求處理，請直接從 azure.ai.agentserver.invocations 套件使用 InvocationAgentServerHost 並實作您自己的叫用處理常式：

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["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()

完整的 Invocations 部署，請參閱 Foundry 託管的 Telegram 範例。 它將 API 管理置於託管代理 webhook 前方，並使用受管理身份、金鑰保存庫 與 Cosmos DB 來保存持久的對話歷史。