Microsoft Foundry 에이전트 서비스에서 호스트된 에이전트를 사용하면 컨테이너화된 에이전트 애플리케이션을 Microsoft 관리형 인프라에 배포할 수 있습니다. 플랫폼은 스케일링, 세션 상태 지속성, 보안 및 수명 주기 관리를 처리하므로 에이전트의 논리에 집중할 수 있습니다. Microsoft Foundry 호스트 에이전트는 일반적으로 사용 가능하며 사용자 고유의 코드 또는 기본 에이전트 프레임워크를 사용하여 빌드된 에이전트를 지원합니다. 이 문서에서는 특히 통합을 호스팅하는 에이전트 프레임워크에 대해 설명합니다.
에이전트 프레임워크 호스팅 통합을 사용하여 최소한의 코드로 Foundry 응답 또는 호출 프로토콜을 통해 Agent를 노출할 수 있습니다. Python 또한 네이티브 Workflow를 에이전트로 변환하지 않고 직접 호스팅하는 것도 지원합니다.
비고
Azure 개발자 CLI(azd) 워크플로를 사용하여 다른 프레임워크를 사용하여 빌드된 에이전트 코드를 Foundry 호스팅 에이전트에 배포할 수도 있습니다. 프레임워크에 구애받지 않는 개념 및 배포 지침은 호스트된 에이전트란?을 참조하세요. 이 문서의 나머지 부분에는 Agent Framework 통합에 중점을 둡니다.
호스트된 에이전트를 사용하는 경우
원하는 경우 Foundry 호스팅 에이전트를 선택합니다.
- 관리형 인프라 - 컨테이너, 웹 서버 또는 크기 조정 규칙을 직접 구성할 필요가 없습니다.
-
기본 제공 세션 관리 - 플랫폼은
$HOME턴 및 유휴 기간에 걸쳐 파일을 유지 및 업로드합니다. - 전용 에이전트 ID - 배포된 모든 에이전트는 모델, 도구 및 다운스트림 서비스에 대한 보안 액세스를 위해 자체 Entra ID를 가져옵니다.
- OpenAI 호환 엔드포인트 - 클라이언트는 응답 프로토콜을 통해 OpenAI 호환 SDK를 사용하여 에이전트와 상호 작용할 수 있습니다.
관련 시나리오
- 실시간 오디오 에이전트의 경우 서버 쪽 음성 활동 감지, 에코 취소 및 노이즈 감소를 위해 Azure Speech in Foundry Tools(Voice Live)에서 호스트된 에이전트를 사용합니다. 자세한 내용은 호스트된 에이전트에서 Voice Live 사용을 참조하세요.
비고
Python agent-framework-foundry-hosting 통합은 사전 출시 상태입니다. 관리되는 호스팅 서비스인 Microsoft Foundry Hosted Agents를 일반적으로 사용할 수 있습니다.
필수 조건
- Azure 구독
-
Azure AI 에이전트 확장을 사용하는 개발자 CLI(
azd):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 서비스에 전달합니다. 로컬 실행은 해당 플랫폼 컨텍스트를 수신하지 않으므로 애플리케이션은 필요할 때 자체 ID 및 상태 제어를 제공해야 합니다.
응답 프로토콜
응답 프로토콜은 대부분의 에이전트에 권장되는 시작점입니다. 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("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();
Foundry AgentHost.CreateBuilder 호스팅 환경에 대해 미리 구성된 애플리케이션 호스트를 만듭니다.
AddFoundryResponses 는 에이전트를 응답 프로토콜 처리기에 등록하고 MapFoundryResponses HTTP 엔드포인트를 /responses 매핑합니다.
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 응답 프로토콜을 통해 노출합니다. 호출자의 store 필드는 외부 응답과 호스트 관리 세션 및 승인 상태가 저장되는지 여부를 제어합니다. 이 설정은 history_source 모델 기록을 제공하는 사용자를 독립적으로 선택합니다.
history_source |
모델 기록 동작 |
|---|---|
"agent_server"(기본값) |
호스트는 저장된 외부 응답 기록을 재구성하고 중복 기록을 방지하기 위해 다운스트림 서비스 스토리지를 사용하지 않도록 설정합니다. |
"service" |
호스트는 현재 입력만 보내고 저장 모델 서비스의 연속 ID를 비공개로 저장합니다. 저장된 공급자 대화는 이전 응답에서 분기할 수 없습니다. |
"agent" |
호스트는 현재 입력만 보냅니다. 에이전트의 HistoryProvider 또는 다운스트림 서비스 스토리지의 기본 설정이 이력 관리를 담당합니다. |
부하가 활성화된 "agent_server"와 HistoryProvider 또는 "service"을 함께 결합하지 마세요. 기본 모드는 고정 다운스트림 연속 옵션(예: conversation_id, previous_response_id및 conversation.)도 거부합니다. 사용자 지정 history_source="agent" 구현에는 SupportsAgentRun를 사용합니다.
생성자 매개 변수는 response_store 외부 응답 지속성에 대한 백 엔드를 선택합니다. 이전 생성자 매개 변수 store 는 더 이상 사용되지 않는 별칭 response_store입니다. 두 매개 변수 모두 호출자의 요청 store 별 필드를 설정하지 않습니다. 요청 store=false 은 일회성입니다. 호스트 관리 상태를 저장하지 않고 지원되는 다운스트림 스토리지를 사용하지 않도록 설정하며 사용할 background=true수 없습니다.
호스트는 제공된 에이전트를 소유하며 호스팅 관련 컨텍스트 공급자를 추가할 수 있습니다. 에이전트를 다른 호스트와 함께 다시 사용하거나 호스트 생성 후 직접 호출하지 마세요.
응답 호스트는 네이티브 컴퓨터 호출, 스크린샷 및 안전 검사를 유지합니다. 애플리케이션은 요청된 작업을 실행하고 안전 검사를 명시적으로 승인해야 합니다. 전체 흐름은 네이티브 컴퓨터 사용을 참조하세요.
에이전트 인스턴스 또는 팩터리 선택
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 세션, 검사점 및 함수 승인 저장소를 통해 지원되는 상태를 계속합니다. 요청 범위 에이전트에 임의의 필드를 유지하지 않습니다.
워크플로 및 복원력 있는 장기 실행 워크플로 샘플을 참조하세요.
또한 통합이 요청 ID를 전달하거나 요청별 리소스를 소유하는 경우 팩터리를 사용합니다. 예를 들어 현재 플랫폼 호출 또는 사용자 컨텍스트를 사용할 때 MCP 연결, 도구 상자, 기술 공급자, 검색 클라이언트, 메모리 공급자 및 해당 자격 증명을 팩터리 내에 만듭니다. 프로세스 전체 MCP 연결을 다시 사용하면 해당 연결을 연 요청의 ID를 유지할 수 있습니다.
호스트는 각 요청에 대해 팩터리에서 만든 에이전트를 입력하고 종료합니다.
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]을 허용하는 레거시 워크플로의 경우 현재 Responses 턴만 변환하려면 response_input_messages(request)를 사용합니다.
이전 외부 기록을 로드하거나 보류 중인 워크플로 회신을 디코딩하지 않습니다. 호스팅 agent=workflow.as_agent() 은 현재 베타 중에 계속 사용할 수 있지만 사용 중단 경고를 내보낸다. 전체 예제는 네이티브 Responses 워크플로 샘플을 참조하세요.
상태 저장 및 장시간 실행되는 대화 처리
ResponsesHostServer 및 InvocationsHostServer 기본적으로 영구 세션 저장소를 구성합니다. 는 를 제공합니다. 응답 세션은 논리 저장소를 사용하고, 호출 세션은 별도의 저장소를 사용합니다. 이러한 저장소는 호스팅될 때 Foundry 상태 저장소를 사용하고 로컬로 실행할 때 SDK의 파일 지원 스토리지를 사용합니다.
응답 워크플로 에이전트의 경우, CheckpointStoreProvider는 FoundryCheckpointStore를 제공합니다. 네이티브 응답 및 호출 워크플로는 정확히 일치하는 연속 체크포인트에 동일한 프로바이더를 사용합니다.
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"를 사용하면 구성된 세션 저장소는 InMemoryHistoryProvider에서 전달되는 공급자 상태(여기에는 AgentSession의 메시지도 포함됨)를 유지합니다.
두 호스트 모두 agent_session_store_provider부터 StoreProvider[SessionStore]까지 허용합니다. 세션 상태는 AgentSession 직렬화를 지원해야 합니다. 사용자 지정 상태 형식에 대한 코덱을 register_state_type()등록합니다. 복원된 상태는 Python 개체 ID를 유지하지 않습니다. 새 기본 저장소는 마지막 쓰기 후 30일 후에 세션이 만료됩니다.
사용자 지정 공급자는 자체 보존을 제어합니다.
범위가 지정된 기본 저장소는 레거시 범위 미지정 agent_sessions, invocation_sessions, 체크포인트 또는 함수 승인 데이터를 읽지 않습니다. 이전 previous_response_id 또는 대화 ID를 다시 사용하는 대신 새 응답 대화를 시작합니다. 호출은 범위가 지정된 저장소의 빈 에이전트 프레임워크 세션으로 시작됩니다.
로드된 레코드는 AgentSession ETag 조건을 사용합니다. 다른 요청이 동일한 세션을 먼저 진행해 세션 상태를 앞서 변경한 경우, 오래된 쓰기는 최신 상태를 덮어쓰는 대신 실패합니다. 이 검사는 에이전트 또는 도구 부작용에 대한 트랜잭션 또는 정확히 한 번 실행을 제공하지 않으므로 애플리케이션은 여전히 겹치는 요청을 조정해야 합니다.
Responses별 저장소의 경우 function_approval_store_provider를 ContextScopedStoreProvider에 전달하거나 StoreProvider를 checkpoint_store_provider에 전달하세요.
외부 백그라운드 작업은 폴링에 호출자에게 표시되는 response.id를 사용합니다. 기본값 background_source="agent_server" 은 호스트에서 백그라운드 실행을 유지합니다.
background_source="provider" 및 저장 및 재개 가능한 Responses 클라이언트와 함께 history_source="service"만 설정하세요.
ResponsesServerOptions(resilient_background=True)도 설정된 경우 호스트는 비공개 연속 토큰을 저장한 후에만 공급자 폴링을 다시 시작할 수 있습니다. 다음 토큰이 저장되기 전에 충돌이 발생하면 로컬 도구의 부작용이 다시 실행될 수 있으므로, 해당 부작용이 멱등성을 갖도록 합니다.
ResponsesHostServer에서 ResponsesServerOptions를 가져와 azure.ai.agentserver.responses 매개변수를 통해 options에 전달합니다. 사용 가능한 장기 실행 대화 옵션은 에이전트 유형에 따라 달라집니다.
| Capability | 에이전트 유형 | 요구 사항 및 동작 |
|---|---|---|
| 워크플로 체크포인트 백그라운드 복구 | 워크플로만 |
ResponsesServerOptions(resilient_background=True)을 설정합니다. 및 와 함께 Responses 요청을 보냅니다. 다시 시작한 후 호스트는 최신 지속성 워크플로 검사점을 다시 시작하거나 검사점이 없는 경우 원래 입력을 재생합니다. 호스트가 관리하므로 워크플로에서 검사점 스토리지를 구성하지 마세요. 마지막 영구 체크포인트 이후의 작업이 반복될 수 있으므로 외부 부작용이 멱등성을 갖도록 하세요. |
| 공급자 기본 백그라운드 응답 | Responses를 저장하는 클라이언트가 있는 워크플로가 아닌 Agent |
설정 history_source="service" 및 background_source="provider". 저장된 공급자 연속 토큰이 호스트 다시 시작 후에도 유지되어야 하는 경우 resilient_background=True로 설정합니다. |
| 조종 가능한 대화 | 일시적으로 사용할 수 없음 |
steerable_conversations=True를 설정하지 마십시오. 호스트는 에이전트 서버 SDK가 거부된 스티어링 턴을 안전하게 처리할 수 있을 때까지 구성 중에 RuntimeError를 발생시킵니다. |
전체 구현은 사용자 지정 스토리지, 기본 응답 기록 및 배경 및복원력 있는 장기 실행 워크플로 샘플을 참조하세요.
호스트된 샌드박스에서 파일 읽기
호스트된 샌드박스의 영구 $HOME 를 일반적인 파일 시스템 경계가 아닌 요청 라우팅 리소스로 처리합니다. 애플리케이션이 명시적으로 전용 디렉터리에 업로드하는 파일만 허용하고, 현재 샌드박스 ID의 유효성을 검사하고, 절대 경로, 순회, 링크, 비정형 파일 및 대형 또는 잘못된 콘텐츠를 거부합니다.
응답 프로토콜의 경우 본문 필드가 있는 호스트된 세션으로 agent_session_id 요청을 라우팅합니다. 쿼리 문자열 선택기는 호출용입니다.
세션 업로드 및 도구 상자 코드 인터프리터 파일은 별도의 리소스입니다. 업로드된 샌드박스 파일은 도구 상자 컨테이너에 자동으로 탑재되지 않습니다.
바인딩된 UTF-8 읽기 및 로컬 및 호스트된 업로드 지침은 세션 파일 샘플을 참조하세요.
요청 옵션 제어
호스트는 네이티브 응답 생성 필드를 Agent Framework 실행 옵션에 매핑합니다. 예를 들어, max_output_tokens는 max_tokens가 되고 parallel_tool_calls은 allow_multiple_tool_calls이 됩니다.
extra_body의 평면화된 값은 번역된 네이티브 값을 재정의합니다.
일반 에이전트가 실행되기 전에 동기 또는 비동 prepare_options(request, options) 기 후크를 사용하여 호출자 모델 옵션을 제거하거나 대체합니다. 후크는 호스트 제어 ID, 스토리지, 연속 또는 전송 필드를 설정할 수 없습니다. 런타임 모델 옵션을 허용할 수 없는 사용자 지정 unsupported_options 구현의 경우 "warn"을 SupportsAgentRun(기본값), "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 합니다.
간단한 설정의 경우 패키지에서 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 는 응답 호스트에 대해 설명된 동일한 인스턴스 또는 요청 범위 팩터리 양식을 허용합니다. 구성된 저장소에서 직렬화된 세션을 복원하므로 호스트가 다시 시작되면 완료된 대화를 계속할 수 있습니다. 스토리지 동작, 보존 및 사용자 지정은 상태 유지 및 장기 실행 대화 처리를 참조하세요.
호스팅되는 경우 호출은 지속 상태에 설명된 확인된 요청 범위를 사용하고 장기 실행 대화를 처리합니다.
AgentSession.session_id를 하나의 불투명한 값으로 취급하세요. 내부 표현을 파싱하거나 이에 의존하지 마세요. 로컬 실행은 기존 단일 사용자 스토리지 동작을 유지합니다.
Invocations를 사용해 네이티브 워크플로를 호스팅하기
parse_request 및 명시적 workflow= 콜백을 전달하여 네이티브 워크플로를 호스팅합니다. 콜백은 애플리케이션 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 예제는 네이티브 호출 워크플로 샘플을 참조하세요.
호출 요청 및 응답 사용자 지정
기본적으로 문자열message, POST /invocations 선택적 개체 및 선택적 options 부울 stream 값이 있는 JSON 개체를 허용합니다. 애플리케이션별 페이로드를 허용하려면 InvocationRun(messages, options, stream)를 반환하는 동기 또는 비동기 parse_request 콜백을 전달합니다. 에이전트를 실행하기 전에 호출자 생성 옵션의 복사본을 필터링하거나 바꾸는 데 사용합니다 prepare_options .
호스트는 후크 출력의 유효성을 검사하고 플랫폼 ID, 스토리지, 연속 및 에이전트 실행 컨트롤을 거부합니다. 런타임 옵션을 허용하지 않는 에이전트의 경우 unsupported_options를 "ignore"(기본값), "error", 또는 "warn"로 설정합니다. 전체 구현은 호출 파서 샘플을 참조하세요.
비 스트리밍 성공은 형식 {"response": "..."}으로 JSON을 반환합니다.
스트리밍은 서버 전송 이벤트를 사용합니다. 즉, 하나 이상의 event: delta 프레임 다음에, 성공 시 event: done, 실패 시 event: error가 이어집니다. 스트림은 오류가 발생하기 전에 델타를 전송할 수 있으므로 클라이언트는 델타가 아니라 done를 성공적인 완료로 처리해야 합니다. 호스트는 응답 스트림을 최종 확정하고 AgentSession를 영구 저장한 후에만 done를 내보냅니다. 해당 session_id은 직렬화된 AgentSession.session_id가 아니라 플랫폼 샌드박스 경로 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["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 Management를 배치하고 지속적인 대화 기록에 관리 ID, Key Vault 및 Cosmos DB를 사용합니다.
비고
Foundry 호스팅 에이전트에 대한 Go 지원이 곧 제공될 예정입니다. 최신 상태는 에이전트 프레임워크 Go 리포지토리 를 참조하세요.
Tip
호스트된 에이전트 프로젝트의 예제는 Python 샘플 또는 C# 샘플을 참조하세요. 또는 명령을 azd ai agent init 사용하여 호스트된 새 에이전트 프로젝트를 처음부터 스캐폴드합니다. 단계별 지침 은 이 빠른 시작 가이드 를 참조하세요.
로컬로 실행
Azure 개발자 CLI(azd)는 호스트된 에이전트를 로컬로 실행하고 테스트하는 가장 쉬운 방법을 제공합니다.
프로젝트를 초기화하기
새 폴더를 만들고 샘플 매니페스트에서 초기화합니다.
mkdir my-hosted-agent && cd my-hosted-agent
azd ai agent init -m <path-to-agent.manifest.yaml>
Tip
매니페스트는 로컬 YAML 파일의 경로이거나 원격 매니페스트의 URL일 수 있습니다.
환경 변수 설정
export FOUNDRY_PROJECT_ENDPOINT="https://<account>.services.ai.azure.com/api/projects/<project>"
export AZURE_AI_MODEL_DEPLOYMENT_NAME="<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에 배포합니다.
리소스 프로비전 (Foundry 프로젝트가 아직 없는 경우):
azd provisionFoundry 인스턴스, 프로젝트, 모델 배포, Application Insights 및 컨테이너 레지스트리를 사용하여 리소스 그룹을 만듭니다.
에이전트를 배포합니다.
azd deploy이렇게 하면 에이전트를 컨테이너 이미지로 패키지하고, Azure Container Registry 푸시하고, Foundry 에이전트 서비스에 배포합니다.
Foundry 호스팅 인프라는 런타임에 에이전트 컨테이너에 다음 환경 변수를 자동으로 삽입합니다.
| Variable | 설명 |
|---|---|
FOUNDRY_PROJECT_ENDPOINT |
Foundry 프로젝트의 엔드포인트 URL입니다. |
AZURE_AI_MODEL_DEPLOYMENT_NAME |
모델 배포 이름(azd ai agent init에서 구성됨)입니다. |
APPLICATIONINSIGHTS_CONNECTION_STRING |
Application Insights를 통한 원격 분석을 위한 연결 문자열. |
배포되면 에이전트는 전용 Foundry 엔드포인트를 통해 액세스할 수 있으며 Foundry 포털에서 테스트할 수도 있습니다.