Agentes Alojados da Foundry

• SDK do .NET 10 ou posterior

Instale o pacote NuGet de alojamento:

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

• Python 3.10 ou posterior

Instale o pacote de alojamento pré-lançamento, o cliente Foundry e o pacote de autenticação 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("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();

O AgentHost.CreateBuilder cria um anfitrião de aplicação pré-configurado para o ambiente de alojamento da Foundry. AddFoundryResponses regista o seu agente com o handler do protocolo Responses e MapFoundryResponses mapeia o /responses endpoint 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()

Encapsula ResponsesHostServer o seu agente, expondo-o através do protocolo Foundry Responses. O campo store do chamador controla se a resposta externa e os estados da sessão e da aprovação geridos pelo host são guardados. O history_source cenário seleciona de forma independente quem fornece a história do modelo:

Não combine "agent_server" ou "service" com um HistoryProvider ativado para carga. O modo predefinido também rejeita opções fixas de continuação a jusante como conversation_id, previous_response_id, e conversation. Use history_source="agent" para uma implementação personalizada de SupportsAgentRun.

O parâmetro do construtor response_store seleciona o backend para a persistência externa de Responses. O parâmetro mais antigo do construtor store é um sinónimo preterido de response_store; nenhum dos parâmetros define o campo store por pedido do autor da chamada. Um pedido com store=false é de utilização única: não guarda o estado gerido pelo anfitrião, desativa o armazenamento suportado em nível inferior e não pode usar background=true.

O host é proprietário do agente fornecido e pode adicionar fornecedores de contexto específicos para o alojamento. Não reutilize o agente com outro host nem o invoque diretamente após a construção do host.

O anfitrião do Responses preserva chamadas nativas do sistema, capturas de ecrã e verificações de segurança. A sua candidatura deve executar as ações solicitadas e reconhecer explicitamente quaisquer verificações de segurança. Para o fluxo completo, veja Uso nativo de computadores.

Escolha uma instância de agente ou fábrica

Tanto ResponsesHostServer como InvocationsHostServer aceitam, através do parâmetro agent, quer uma instância de agente quer um invocável síncrono ou assíncrono sem argumentos. O hospedeiro reutiliza uma instância ao longo de todo o seu ciclo de vida. Uma função invocável é executada uma vez por pedido, e o agente devolvido pertence a esse pedido.

Use um invocável quando um agente regular mantém um estado mutável fora de AgentSession. Os hosts mantêm apenas os seus armazenamentos suportados de sessão, checkpoint e aprovação de funções, não campos arbitrários num agente no âmbito do pedido.

Também use uma fábrica quando uma integração transporta identidade de pedido ou possui recursos específicos de pedido. Por exemplo, crie ligações MCP, caixas de ferramentas, fornecedores de competências, clientes de pesquisa, fornecedores de memória e as respetivas credenciais na fábrica quando utilizam a chamada atual da plataforma ou o contexto do utilizador. Reutilizar uma conexão MCP partilhada por todo o processo pode preservar a identidade do pedido que a abriu.

O anfitrião entra e sai agentes criados em fábrica para cada pedido. Agent gere clientes geridos pelo contexto e ferramentas MCP, mas a sua fábrica deve fechar qualquer outro fornecedor, transporte ou credencial que crie. Não fechem objetos partilhados que a aplicação forneceu de fora da fábrica.

Executar um fluxo de trabalho nativo com Responses

Python pode alojar um fluxo de trabalho construído diretamente através de workflow=. Um fluxo de trabalho nativo requer um parse_response callback que mapeie o pedido de Respostas atual para uma entrada inicial tipada ou para o lote completo de respostas pendentes:

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__}"],
    ),
)

Use uma fábrica síncrona ou assíncrona sensível ao pedido para fluxos de trabalho que possam pausar, continuar ou recuperar trabalho em segundo plano. A fábrica deve devolver um grafo recém-construído com novos executores mutáveis, agentes, clientes, fornecedores e ferramentas. Mantenha o nome do fluxo de trabalho e os IDs do executor estáveis para que o anfitrião possa restaurar o checkpoint exato associado à resposta externa.

O utilizador Trusted Platform e o sandbox da Foundry isolam o estado nativo do fluxo de trabalho. O host valida um lote completo de resposta pendente antes de consumir qualquer permissão de resposta. Respostas obsoletas, parciais, duplicadas, repetidas, cross-user e cross-sandbox falham antes da execução do fluxo de trabalho. Um pedido com store=false não guarda o estado do fluxo de trabalho e não pode devolver uma pausa retomável.

Para um fluxo de trabalho legado que aceita list[Message], use response_input_messages(request) para converter apenas o turno atual de Respostas. Não carrega o outer history anterior nem descodifica respostas pendentes do fluxo de trabalho. A hospedagem agent=workflow.as_agent() permanece disponível durante a beta atual, mas emite um aviso de descontinuação. Para exemplos completos, consulte os exemplos nativos de fluxo de trabalho Responses.

Persistir o estado e lidar com conversas de longa duração

ResponsesHostServer e InvocationsHostServer configuram armazenamentos de sessão persistentes por defeito. AgentSessionStoreProvider fornece um FoundryAgentSessionStore; as sessões Responses usam o arquivo lógico agent_sessions, enquanto as sessões Invocations usam o arquivo separado invocation_sessions. Estes armazenamentos utilizam o Foundry State Store quando estão alojados e o armazenamento baseado em ficheiros do SDK quando são executados localmente.

Para agentes do fluxo de trabalho Responses, CheckpointStoreProvider fornece um FoundryCheckpointStore. Os fluxos de trabalho de Respostas Nativas e Invocações usam o mesmo fornecedor para os seus checkpoints de continuação exatos. FunctionApprovalStoreProvider fornece um FoundryFunctionApprovalStore para aprovações pendentes de ferramentas de agentes. Pedidos nativos de fluxo de trabalho e respostas de aprovação estão, em vez disso, ligados a pontos de verificação do fluxo de trabalho.

Quando é executado no Foundry, o Python padrão armazena o estado do namespace pelo ID do utilizador da plataforma e pelo ID da sessão sandbox do Foundry. Também exigem um ID de chamada de plataforma para cada operação estadual. O ID da chamada autoriza e correlaciona a operação; não é um ID de conversa nem faz parte da chave de armazenamento.

Nas respostas, o FOUNDRY_AGENT_SESSION_ID configurado na plataforma identifica a sandbox, e um agent_session_id diferente fornecido pelo chamador é rejeitado. Para Invocações, o anfitrião verifica o parâmetro de consulta agent_session_id encaminhado em relação ao contexto da solicitação. Se FOUNDRY_AGENT_SESSION_ID não estiver configurado, o parâmetro de consulta deve estar presente, não vazio e corresponder ao contexto do pedido. Valores em falta, duplicados ou em conflito são rejeitados, em vez de se utilizar um ID alternativo do SDK.

Estas garantias aplicam-se às lojas alojadas por predefinição. Os fornecedores de lojas personalizadas devem implementar isolamento equivalente de utilizador e sandbox, preservar o interior AgentSession.session_id separadamente das chaves de pesquisa do host e usar escritas condicionais para que pedidos obsoletos não possam sobrescrever snapshots mais recentes. As novas chaves devem usar operações de escrita apenas de criação, em vez de upserts incondicionais. Consulte o exemplo de armazenamento personalizado para uma implementação do Cosmos DB com escritas e eliminações protegidas por ETag.

Com history_source="agent", o armazenamento de sessões configurado persiste o estado do fornecedor transportado por AgentSession, incluindo as mensagens de InMemoryHistoryProvider.

Ambos os anfitriões aceitam StoreProvider[SessionStore] através de agent_session_store_provider. O estado da sessão deve suportar a AgentSession serialização. Regista codecs para tipos de estado personalizados com register_state_type(); o estado restaurado não preserva a identidade dos objetos Python. Os novos armazenamentos predefinidos fazem expirar as sessões 30 dias após a última gravação. Os provedores personalizados gerem a sua própria política de retenção.

Os repositórios predefinidos com escopo não leem dados legados sem escopo agent_sessions, invocation_sessions, de checkpoint nem de aprovação de funções. Inicie uma nova conversa do Responses em vez de reutilizar um previous_response_id ou ID de conversa antigo. As invocações começam com uma sessão vazia do Agent Framework no armazenamento com âmbito definido.

Os registos carregados AgentSession usam condições ETag. Se outro pedido fizer avançar primeiro a mesma sessão, a escrita desatualizada falha em vez de substituir o estado mais recente. Esta verificação não fornece transações nem execução exata uma vez para efeitos secundários do agente ou da ferramenta, pelo que as aplicações têm de coordenar pedidos sobrepostos.

Para armazenamento específico de Respostas, passe a StoreProvider para function_approval_store_provider ou a ContextScopedStoreProvider para checkpoint_store_provider.

O trabalho em segundo plano externo utiliza o response.id visível para o chamador para consulta. O padrão background_source="agent_server" mantém a execução em segundo plano no host. Defina background_source="provider" apenas com history_source="service" e um cliente Responses com armazenamento e capacidade de retoma. Se ResponsesServerOptions(resilient_background=True) também estiver definido, o host só pode retomar a sondagem do fornecedor depois de guardar o token de continuação privado. Torna os efeitos secundários da ferramenta local idempotentes, porque uma falha antes de o próximo token ser guardado pode repeti-los.

Importa ResponsesServerOptions de azure.ai.agentserver.responses, e passa para ResponsesHostServer através do options parâmetro. As opções de conversa de longa duração disponíveis dependem do tipo de agente:

Para implementações completas, consulte os exemplos de armazenamento personalizado, histórico e antecedentes básicos de Respostas, e fluxos de trabalho resilientes e de longa duração .

Leia ficheiros do sandbox alojado

Trate o $HOME persistente de uma sandbox alojada como um recurso encaminhado com base em pedidos, e não como um limite genérico do sistema de ficheiros. Aceite apenas ficheiros que a sua aplicação carregue explicitamente para um diretório dedicado, valide a identidade atual do sandbox e rejeite caminhos absolutos, percursos, ligações, ficheiros não regulares e conteúdos sobredimensionados ou inválidos.

Para o protocolo Responses, encaminhe uma solicitação para uma sessão alojada através do campo agent_session_id no corpo do pedido. O seletor de sequência de consulta é para Invocações. Os carregamentos de sessão e os ficheiros de interpretador de código Toolbox são recursos separados; um ficheiro sandbox carregado não é automaticamente montado num contentor Toolbox. Consulte o exemplo de ficheiros de sessão para leituras UTF-8 limitadas e orientações de upload local e alojado.

Opções de pedido de controlo

O host mapeia campos nativos de geração de Respostas para opções de execução do Agent Framework. Por exemplo, max_output_tokens torna-se max_tokens, e parallel_tool_calls torna-se allow_multiple_tool_calls. Os valores nivelados de extra_body sobrepõem-se aos valores nativos traduzidos.

Utilize o hook síncrono ou assíncrono prepare_options(request, options) para remover ou substituir opções do modelo do chamador antes de um agente normal ser executado. O hook não pode definir campos de identidade, armazenamento, continuação ou transporte controlados pelo hospedeiro. Para uma implementação personalizada SupportsAgentRun que não aceite opções de modelo em tempo de execução, defina unsupported_options para "warn" (o padrão), "ignore", ou "error".

Processar pedidos de consentimento do OAuth

Quando uma ferramenta MCP alojada no Foundry requer consentimento do utilizador, ResponsesHostServer devolve uma resposta incompleta com um oauth_consent_request item de saída. Apresente-o ao utilizador consent_link e, em seguida, continue com o ID da resposta incompleta como previous_response_id depois de o utilizador dar o seu consentimento. O anfitrião mantém a sessão do agente para esta nova tentativa e disponibiliza apenas hiperligações de consentimento HTTPS absolutas.

Se o seu anfitrião souber a origem esperada da autorização, restrinja os links de consentimento com allowed_oauth_consent_origins:

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

Omitir a lista de permissões mantém a validação absoluta de HTTPS sem restringir a origem de destino. Fornecer uma lista vazia rejeita todos os links de consentimento. Configure apenas origens HTTPS exatas; as entradas que incluam um caminho, parâmetros de consulta ou fragmento são rejeitadas.

Com o protocolo Invocations em C#, implementa-se um personalizado InvocationHandler para processar os pedidos recebidos.

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();

O AddInvocationsServer método regista os serviços do protocolo Invocations. Implementa InvocationHandler para definir como o seu agente processa cada pedido.

Para uma configuração leve, use InvocationsHostServer do pacote agent_framework_foundry_hosting. Envolve o seu agente de forma semelhante ao ResponsesHostServer e gere as sessões automaticamente:

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 aceita a mesma instância ou as formas de fábrica com âmbito de pedido descritas para o host Responses. Restaura sessões serializadas do armazenamento configurado, permitindo que as conversas em curso continuem depois de o anfitrião reiniciar. Para informações sobre o comportamento do armazenamento, a retenção e a personalização, consulte Persistir o estado e gerir conversas de longa duração.

Quando alojado, o Invocations utiliza o âmbito de pedido verificado descrito no estado Persist e gere conversas de longa duração. Trate AgentSession.session_id como um valor opaco; não analise nem dependa da sua representação interna. As execuções locais mantêm o comportamento existente de armazenamento de utilizador único.

Alojar um fluxo de trabalho nativo com Invocações

Passe workflow= e uma chamada de retorno explícita parse_request para alojar um fluxo de trabalho nativo. O callback detém o esquema JSON da aplicação e devolve um WorkflowTurn com entrada tipada ou com o lote completo de resposta pendente:

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__}",
        ],
    ),
)

Inclua todos os tipos de aplicação personalizada que o fluxo de trabalho guarda na lista de allowed_checkpoint_types do fornecedor de checkpoints.

Workflows alojados requerem uma fábrica sensível ao pedido que devolve um gráfico recém-construído com IDs de workflow e de executor estáveis. Um fluxo de trabalho integrado direto está disponível apenas para uma execução local, one-shot, que não pausa.

As respostas de workflow não em streaming utilizam JSON da aplicação com uma lista de eventos output. O streaming emite eventos enquadrados output e request_info, seguidos de done apenas depois de o cursor exato do fluxo de trabalho ser guardado. Trate a saída transmitida como provisória até done. Os fluxos de trabalho nativos não suportam legacy_wire_format=True.

O host valida as respostas em relação ao checkpoint exato pendente no âmbito do utilizador confiável e do sandbox. Se um fluxo de trabalho tiver vários pedidos pendentes, responda ao lote completo numa única interação. Para um parser executável, workflow de tickets tipados, lista de permissões de tipos de checkpoint e exemplos de JSON/SSE, consulte o exemplo de workflow nativo Invocations.

Personalizar pedidos e respostas de invocações

Por defeito, POST /invocations aceita um objeto JSON com uma string message, um objeto opcional options e um valor Booleano stream opcional. Para aceitar uma carga útil específica da aplicação, passe um callback síncrono ou assíncrono parse_request que devolve InvocationRun(messages, options, stream). Use prepare_options para filtrar ou substituir uma cópia das opções de geração de chamadas antes da execução do agente.

O host valida a saída do hook e rejeita a identidade da plataforma, armazenamento, continuação e controlos de execução do agente. Para agentes que não aceitam opções de tempo de execução, defina unsupported_options para "warn" (o padrão), "ignore", ou "error". Consulte o exemplo do parser Invocations para uma implementação completa.

Em caso de sucesso sem streaming, retorna JSON sob a forma de {"response": "..."}. O streaming utiliza eventos enviados pelo servidor: um ou mais event: delta frames, seguidos de event: done em caso de sucesso ou event: error de falha. Um fluxo pode emitir deltas antes de ocorrer um erro, pelo que os clientes devem considerar done, e não um delta, como indicação de conclusão com êxito. O anfitrião emite done apenas depois de finalizar o fluxo de resposta e persistir o AgentSession. O seu session_id é o ID da rota sandbox da plataforma, não o AgentSession.session_id serializado.

Defina legacy_wire_format=True apenas durante a migração de clientes existentes que necessitam da resposta anterior em texto simples e do fluxo bruto de segmentos de texto. Este modo de compatibilidade está obsoleto e não converte falhas em texto bem-sucedido. O host serializa pedidos na mesma sessão apenas dentro de um processo; um conflito de comparação e troca entre processos pode ainda ocorrer após efeitos de ferramenta externa.

O protocolo Invocations não retoma execuções de workflow pendentes ou interrompidas. Use o padrão de handler personalizado na secção seguinte quando precisar de um comportamento diferente de continuação do fluxo de trabalho.

Para controlo total sobre o tratamento dos pedidos, use InvocationAgentServerHost diretamente do azure.ai.agentserver.invocations pacote e implemente o seu próprio handler de invocação:

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

Para uma implementação completa do Invocations, consulte o exemplo do Telegram hospedado pela Foundry. Coloca a API Management à frente do webhook do agente hospedado e utiliza identidades geridas, Key Vault e Cosmos DB para um histórico duradouro de conversas.