Agentes hospedados da Foundry

• SDK do .NET 10 ou posterior

Instale o pacote NuGet de hospedagem:

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

• Python 3.10 ou posterior

Instale o pacote de hospedagem de pré-lançamento, o cliente Foundry e o pacote de autenticação do 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 host de aplicativo pré-configurado para o ambiente de hospedagem do Foundry. AddFoundryResponses registra seu agente com o manipulador de 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()

O ResponsesHostServer encapsula seu agente e o expõe por meio do protocolo Foundry Responses. O campo do store chamador controla se a resposta externa e a sessão gerenciada pelo host e o estado de aprovação são salvos. A history_source configuração seleciona independentemente quem fornece o histórico do modelo:

Não combine "agent_server" ou "service" com HistoryProvider com suporte a carga. O modo padrão também rejeita opções de continuação downstream fixas, como conversation_id, previous_response_ide conversation. Use history_source="agent" para uma implementação SupportsAgentRun personalizada.

O parâmetro do construtor response_store define o backend para a persistência externa de Responses. O parâmetro antigo do construtor store é um alias obsoleto para response_store; nenhum dos parâmetros define o campo store do chamador para cada solicitação. Uma solicitação com store=false é de uso único: ela não salva o estado gerenciado pelo host, desativa o armazenamento downstream compatível e não pode usar background=true.

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

O host do Responses preserva chamadas nativas do computador, capturas de tela e verificações de segurança. Seu aplicativo deve executar as ações solicitadas e reconhecer explicitamente todas as verificações de segurança. Para obter o fluxo completo, consulte o uso do computador nativo.

Escolha uma instância ou fábrica de agentes

Tanto InvocationsHostServer quanto ResponsesHostServer aceitam uma instância de agente ou um callable síncrono ou assíncrono sem argumentos por meio do parâmetro agent. O host reutiliza uma instância durante todo o seu ciclo de vida. Um callable é executado uma vez por requisição, e o agente retornado pertence a essa requisição.

Use um callable quando um agente regular retém o estado mutável fora de AgentSession. Os hosts persistem apenas os repositórios com suporte de sessão, ponto de verificação e aprovação de função, não campos arbitrários em um agente com escopo de solicitação.

Use também uma fábrica quando uma integração carrega a identidade da solicitação ou possui recursos específicos da solicitação. Por exemplo, crie conexões MCP, Toolboxes, provedores de habilidades, clientes de pesquisa, provedores de memória e suas credenciais dentro da fábrica quando eles usarem a chamada de plataforma atual ou o contexto do usuário. Reutilizar uma conexão MCP em todo o processo pode manter a identidade da solicitação que a abriu.

O host entra e sai dos agentes criados pela fábrica para cada solicitação. Agent gerencia clientes gerenciados por contexto e ferramentas MCP, mas sua fábrica deve fechar qualquer outro provedor, transporte ou credencial que ela criar. Não feche os objetos compartilhados fornecidos pelo aplicativo de fora da fábrica.

Hospede um fluxo de trabalho nativo com Respostas

O Python pode hospedar um fluxo de trabalho criado diretamente por meio de workflow=. Um fluxo de trabalho nativo requer um parse_response callback que mapeia a solicitação atual de Responses para uma entrada inicial tipada ou 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 com reconhecimento de solicitação para fluxos de trabalho que possam pausar, continuar ou recuperar o trabalho em segundo plano. A fábrica deve retornar um grafo recém-criado com executores, agentes, clientes, provedores e ferramentas mutáveis e atualizados. Mantenha o nome do fluxo de trabalho e as IDs do executor estáveis para que o host possa restaurar o ponto de verificação exato associado à resposta externa.

O usuário de plataforma confiável e o sandbox do Foundry isolam o estado do fluxo de trabalho nativo. O host valida um lote completo de respostas pendentes antes de consumir qualquer autoridade de resposta. Respostas obsoletas, parciais, duplicadas, reproduzidas, entre usuários e entre sandboxes falham antes da execução do fluxo de trabalho. Uma solicitação com store=false não salva o estado do fluxo de trabalho e não pode retornar uma pausa retomável.

Para um fluxo de trabalho herdado que aceita list[Message], use response_input_messages(request) para converter apenas o turno atual de Responses. Não carrega o histórico externo anterior nem decodifica respostas de fluxo de trabalho pendentes. A hospedagem agent=workflow.as_agent() permanece disponível durante o beta atual, mas emite um aviso de descontinuação. Para obter exemplos completos, consulte os exemplos do fluxo de trabalho native Responses.

Manter o estado e lidar com conversas de longa execução

ResponsesHostServer e InvocationsHostServer configure repositórios de sessão persistentes por padrão. AgentSessionStoreProvider fornece um FoundryAgentSessionStore; as sessões de resposta usam o repositório lógico agent_sessions, enquanto as sessões de invocação usam o repositório separado invocation_sessions. Esses repositórios usam o Repositório de Estado do Foundry quando hospedados e o armazenamento baseado em arquivos do SDK quando executados localmente.

Para agentes de fluxo de trabalho do Responses, CheckpointStoreProvider fornece um FoundryCheckpointStore. Os fluxos de trabalho de Respostas Nativas e Invocações usam o mesmo provedor para seus pontos de verificação de continuação exatos. FunctionApprovalStoreProvider fornece um FoundryFunctionApprovalStore para aprovações pendentes da ferramenta do agente. Em vez disso, as respostas de solicitação e aprovação de fluxo de trabalho nativo são associadas a pontos de verificação de fluxo de trabalho.

Ao executar no Foundry, o padrão Python armazena o estado do namespace pela ID do usuário da plataforma e pela ID da sessão da área restrita do Foundry. Eles também exigem um identificador de chamada da plataforma para cada operação de estado. A ID da chamada autoriza e correlaciona a operação; não é uma ID de conversa e não faz parte da chave de armazenamento.

Para Respostas, o FOUNDRY_AGENT_SESSION_ID configurado pela plataforma identifica o sandbox, e um agent_session_id diferente, fornecido pelo chamador, é rejeitado. Para invocações, o host verifica o parâmetro de consulta encaminhado agent_session_id com base no 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 da solicitação. Valores ausentes, duplicados ou conflitantes são rejeitados em vez de usar uma ID de fallback do SDK.

Essas garantias se aplicam aos repositórios hospedados padrão. Provedores de armazenamento personalizados devem implementar isolamento equivalente de usuário e sandbox, preservar o AgentSession.session_id interno separadamente das chaves de busca do host e usar escritas condicionais para que solicitações desatualizadas não possam sobrescrever instantâneos mais recentes. As novas chaves devem usar escritas do tipo create-only em vez de upserts incondicionais. Consulte o exemplo de armazenamento personalizado para uma implementação do Cosmos DB com gravações e exclusões protegidas por ETag.

Com history_source="agent", o repositório de sessão configurado persiste o estado do provedor transportado por AgentSession, incluindo mensagens de InMemoryHistoryProvider.

Ambos os hosts aceitam de StoreProvider[SessionStore] a agent_session_store_provider. O estado da sessão deve dar suporte à AgentSession serialização. Registre codecs para tipos de estado personalizados com register_state_type(); o estado restaurado não preserva a identidade do objeto em Python. Novos repositórios padrão expiram sessões 30 dias após a última gravação. Os provedores personalizados controlam sua própria retenção.

Os repositórios padrão com escopo não leem dados legados sem escopo agent_sessions, invocation_sessions, de checkpoint ou de aprovação de função. 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 repositório com escopo.

Registros carregados AgentSession usam condições de ETag. Se outra solicitação avançar na mesma sessão primeiro, a gravação obsoleta falhará em vez de substituir o estado mais recente. Essa verificação não fornece transações ou execução exatamente uma vez para efeitos colaterais de agentes ou ferramentas, portanto, os aplicativos ainda devem coordenar solicitações sobrepostas.

Para armazenamento específico para Responses, passe um StoreProvider para function_approval_store_provider ou um ContextScopedStoreProvider para checkpoint_store_provider.

O trabalho em segundo plano externo usa o chamador visível response.id para sondagem. O padrão background_source="agent_server" mantém a execução em segundo plano no host. Defina background_source="provider" somente com history_source="service" e um cliente de Respostas retomável que armazena. Se ResponsesServerOptions(resilient_background=True) também estiver definido, o host poderá retomar a sondagem do provedor somente após salvar o token de continuação privado. Torne os efeitos colaterais da ferramenta local idempotentes porque uma falha antes do próximo token ser salvo pode reproduzi-los.

Importe ResponsesServerOptions de azure.ai.agentserver.responses e passe-o para options por meio do parâmetro ResponsesHostServer. As opções disponíveis de conversa de longa duração dependem do tipo de agente:

Para implementações completas, consulte os exemplos de armazenamento personalizado, histórico básico do Responses e processamento em segundo plano e fluxo de trabalho resiliente de longa duração.

Ler arquivos do sandbox hospedado

Trate o $HOME persistente de uma sandbox hospedada como um recurso roteado por requisição, não como uma fronteira geral do sistema de arquivos. Aceite apenas os arquivos que seu aplicativo carrega explicitamente para um diretório dedicado, valide a identidade atual da sandbox e rejeite caminhos absolutos, travessia de diretórios, links, arquivos não regulares e conteúdo excessivamente grande ou inválido.

Para o protocolo de Respostas, encaminhe uma solicitação para uma sessão hospedada com o campo de corpo agent_session_id. O seletor de string de consulta é para invocações. Os arquivos enviados na sessão e os arquivos do interpretador de código do Toolbox são recursos separados; um arquivo do sandbox enviado não é montado automaticamente em um contêiner do Toolbox. Consulte o exemplo de arquivos de sessão para leituras UTF-8 com limite e orientações sobre uploads locais e hospedados.

Opções de solicitação de controle

O host mapeia campos de geração de respostas nativas para opções de execução do Agent Framework. Por exemplo, max_output_tokens torna-se max_tokense parallel_tool_calls se torna allow_multiple_tool_calls. Valores nivelados de extra_body substituem valores nativos traduzidos.

Use o gancho síncrono ou assíncrono prepare_options(request, options) para remover ou substituir as opções de modelo de chamador antes que um agente regular seja executado. O hook não pode configurar os campos de identidade, armazenamento, continuação ou transporte controlados pelo host. Para uma implementação personalizada SupportsAgentRun que não pode aceitar opções de modelo de runtime, defina unsupported_options como "warn" (o padrão) "ignore"ou "error".

Lidar com solicitações de consentimento do OAuth

Quando uma ferramenta MCP hospedada pela Foundry requer consentimento do usuário, ResponsesHostServer retorna uma resposta incompleta com um oauth_consent_request item de saída. Apresente seu consent_link ao usuário e, em seguida, continue com a ID da resposta incompleta como previous_response_id depois que o usuário fornecer seu consentimento. O host preserva a sessão do agente para essa repetição e expõe apenas links de consentimento HTTPS absolutos.

Se o host souber as origens de autorização esperadas, 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 de HTTPS absoluto sem restringir a origem do destino. Fornecer uma lista vazia rejeita cada link de consentimento. Configure apenas origens HTTPS exatas; entradas que incluam caminho, consulta ou fragmento são rejeitadas.

Com o protocolo Invocações em C#, você implementa uma classe personalizada InvocationHandler para processar solicitações de entrada:

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 método AddInvocationsServer registra os serviços do Protocolo de Invocações. Você implementa InvocationHandler para definir como seu agente processa cada solicitação.

Para uma configuração leve, use InvocationsHostServer do pacote agent_framework_foundry_hosting. Ele encapsula seu agente da mesma forma ResponsesHostServer e manipula o gerenciamento de sessão 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 as mesmas formas de instância ou de fábrica com escopo de solicitação descritas para o host Responses. Ele restaura sessões serializadas do repositório configurado, para que as conversas concluídas possam continuar após a reinicialização do host. Para obter informações sobre comportamento de armazenamento, retenção e personalização, consulte Persistir o estado e lidar com conversas de longa duração.

Quando hospedadas, invocações usam o escopo de solicitação verificado descrito no estado Persist e lidam com conversas de longa execução. Trate AgentSession.session_id como um valor opaco; não analise ou dependa de sua representação interna. As execuções locais mantêm o comportamento de armazenamento de usuário único existente.

Hospede um fluxo de trabalho nativo com Invocações

Passe workflow= e um retorno de chamada parse_request explícito para hospedar um fluxo de trabalho nativo. O retorno de chamada possui o esquema JSON do aplicativo e retorna um WorkflowTurn com entrada digitada ou o lote completo de respostas pendentes.

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 aplicativos personalizados que o fluxo de trabalho salva na lista allowed_checkpoint_types do provedor de ponto de verificação.

Fluxos de trabalho hospedados exigem uma fábrica que reconheça a solicitação e retorne um gráfico recém-construído com IDs de fluxo de trabalho e executor estáveis. Um fluxo de trabalho construído diretamente está disponível apenas para uma execução local única que não pausa.

As respostas de fluxo de trabalho que não são streaming usam o JSON do aplicativo com uma output lista de eventos. O streaming emite eventos output e done delimitados, seguidos por request_info somente após o cursor exato do fluxo de trabalho ser salvo. Trate a saída transmitida como provisória até done. Fluxos de trabalho nativos não dão suporte a legacy_wire_format=True.

O host valida as respostas em relação ao ponto de verificação pendente exato no escopo de usuário confiável e sandbox. Se um fluxo de trabalho tiver várias solicitações pendentes, responda ao lote completo em uma única interação. Para um analisador executável, fluxo de trabalho de tíquete tipado, lista de permissões de tipo de ponto de verificação e exemplos de JSON/SSE, consulte o exemplo de fluxo de trabalho de Invocações nativas.

Personalizar as solicitações e as respostas de Invocations

Por padrão, POST /invocations aceita um objeto JSON com uma cadeia de caracteres message, um objeto opcional options e um valor booliano stream opcional. Para aceitar um payload específico da aplicação, passe um callback síncrono ou assíncrono parse_request que retorne InvocationRun(messages, options, stream). Use prepare_options para filtrar ou substituir uma cópia das opções de geração do chamador antes da execução do agente.

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

Uma resposta bem-sucedida sem streaming retorna um JSON no formato {"response": "..."}. O streaming utiliza eventos enviados pelo servidor: um ou mais frames event: delta, seguidos por event: done em caso de sucesso ou por event: error em caso de falha. Um fluxo pode emitir deltas antes de um erro; portanto, os clientes devem tratar done, e não um delta, como conclusão bem-sucedida. O host emite done somente depois de finalizar o fluxo de resposta e persistir o AgentSession. session_id é o ID da rota de sandbox da plataforma, não o AgentSession.session_id serializado.

Defina legacy_wire_format=True somente ao migrar clientes existentes que exijam a resposta anterior em texto simples e o fluxo bruto de blocos de texto. Esse modo de compatibilidade foi preterido e não converte falhas em texto bem-sucedido. O host serializa solicitações da mesma sessão somente em um processo; um conflito de comparação e troca entre processos ainda pode ocorrer após efeitos de ferramenta externa.

O protocolo Invocações não retoma as execuções de fluxo de trabalho pendentes ou interrompidas. Use o padrão de manipulador personalizado na seção a seguir quando precisar de um comportamento de continuação de fluxo de trabalho diferente.

Para obter controle total sobre o tratamento de solicitações, use InvocationAgentServerHost diretamente do azure.ai.agentserver.invocations pacote e implemente seu próprio manipulador 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 obter uma implantação completa de Invocações, consulte o exemplo do Telegram hospedado pela Foundry. Posiciona o API Management à frente do webhook do agente hospedado e usa identidades gerenciadas, Key Vault e Cosmos DB para um histórico de conversas persistente.