Agent Framework Azure 호스팅 에이전트
Microsoft Agent Framework Python SDK를 사용하여 Azure AI Foundry에서 지속적인 에이전트를 구축합니다.
아키텍처
사용자 쿼리 → AzureAIAgentsProvider → Azure AI Agent Service (지속적)
↓
Agent.run() / Agent.run_stream()
↓
도구: 함수 | 호스팅 (코드/검색/웹) | MCP
↓
AgentThread (대화 지속성)
설치
# 전체 프레임워크 (권장)
pip install agent-framework --pre
# 또는 Azure 전용 패키지만 설치
pip install agent-framework-azure-ai --pre
환경 변수
export AZURE_AI_PROJECT_ENDPOINT="https://<project>.services.ai.azure.com/api/projects/<project-id>" # 모든 인증 방식에 필요
export AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini" # 모든 인증 방식에 필요
export BING_CONNECTION_ID="your-bing-connection-id" # 웹 검색용
export AZURE_TOKEN_CREDENTIALS=prod # 프로덕션에서 DefaultAzureCredential 사용 시에만 필요
인증 및 수명 주기
🔑 아래 모든 코드 샘플에 적용되는 두 가지 규칙:
DefaultAzureCredential을 우선 사용합니다. 로컬(Azure CLI / VS Code / Developer CLI)과 Azure(관리 ID, 워크로드 ID)에서 코드 변경 없이 작동합니다. 연결 문자열이나 계정/API 키는 Entra 감사 및 회전을 우회하므로 사용하지 마십시오.
- 로컬 개발:
DefaultAzureCredential을 그대로 사용합니다.- 프로덕션:
AZURE_TOKEN_CREDENTIALS=prod(또는AZURE_TOKEN_CREDENTIALS=<특정_자격증명>)를 설정하여 자격 증명 체인을 프로덕션에 안전한 자격 증명으로 제한합니다.- 모든 클라이언트를 컨텍스트 관리자로 래핑하여 HTTP 전송, 소켓 및 토큰 캐시가 결정적으로 해제되도록 합니다:
- 동기:
with <Client>(...) as client:- 비동기:
async with <Client>(...) as client:그리고 (azure.identity.aio의)async with DefaultAzureCredential() as credential:코드 조각에서 이 설정을 간소화할 수 있지만, 프로덕션 코드는 항상 두 규칙을 따라야 합니다.
from azure.identity.aio import AzureCliCredential, DefaultAzureCredential, ManagedIdentityCredential
# 개발
credential = AzureCliCredential()
# 프로덕션
# 로컬 개발: DefaultAzureCredential. 프로덕션: AZURE_TOKEN_CREDENTIALS=prod 또는 AZURE_TOKEN_CREDENTIALS=<특정_자격증명> 설정
credential = DefaultAzureCredential(require_envvar=True)
# 또는 프로덕션에서 특정 자격 증명을 직접 사용:
# 참고: https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
핵심 워크플로
기본 에이전트
import asyncio
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="MyAgent",
instructions="You are a helpful assistant.",
)
result = await agent.run("Hello!")
print(result.text)
asyncio.run(main())
함수 도구가 있는 에이전트
from typing import Annotated
from pydantic import Field
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
def get_weather(
location: Annotated[str, Field(description="City name to get weather for")],
) -> str:
"""Get the current weather for a location."""
return f"Weather in {location}: 72°F, sunny"
def get_current_time() -> str:
"""Get the current UTC time."""
from datetime import datetime, timezone
return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="WeatherAgent",
instructions="You help with weather and time queries.",
tools=[get_weather, get_current_time], # 함수를 직접 전달
)
result = await agent.run("What's the weather in Seattle?")
print(result.text)
호스팅 도구가 있는 에이전트
from agent_framework import (
HostedCodeInterpreterTool,
HostedFileSearchTool,
HostedWebSearchTool,
)
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="MultiToolAgent",
instructions="You can execute code, search files, and search the web.",
tools=[
HostedCodeInterpreterTool(),
HostedWebSearchTool(name="Bing"),
],
)
result = await agent.run("Calculate the factorial of 20 in Python")
print(result.text)
스트리밍 응답
async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="StreamingAgent",
instructions="You are a helpful assistant.",
)
print("Agent: ", end="", flush=True)
async for chunk in agent.run_stream("Tell me a short story"):
if chunk.text:
print(chunk.text, end="", flush=True)
print()
대화 스레드
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="ChatAgent",
instructions="You are a helpful assistant.",
tools=[get_weather],
)
# 대화 지속성을 위한 스레드 생성
thread = agent.get_new_thread()
# 첫 번째 턴
result1 = await agent.run("What's the weather in Seattle?", thread=thread)
print(f"Agent: {result1.text}")
# 두 번째 턴 - 컨텍스트 유지됨
result2 = await agent.run("What about Portland?", thread=thread)
print(f"Agent: {result2.text}")
# 나중에 재개할 수 있도록 스레드 ID 저장
print(f"Conversation ID: {thread.conversation_id}")
구조화된 출력
from pydantic import BaseModel, ConfigDict
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
class WeatherResponse(BaseModel):
model_config = ConfigDict(extra="forbid")
location: str
temperature: float
unit: str
conditions: str
async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="StructuredAgent",
instructions="Provide weather information in structured format.",
response_format=WeatherResponse,
)
result = await agent.run("Weather in Seattle?")
weather = WeatherResponse.model_validate_json(result.text)
print(f"{weather.location}: {weather.temperature}°{weather.unit}")
제공자 메서드
| 메서드 | 설명 |
|---|---|
create_agent() | Azure AI 서비스에 새 에이전트 생성 |
get_agent(agent_id) | ID로 기존 에이전트 검색 |
as_agent(sdk_agent) | SDK Agent 객체 래핑 (HTTP 호출 없음) |
호스팅 도구 빠른 참조
| 도구 | 임포트 | 용도 |
|---|---|---|
HostedCodeInterpreterTool | from agent_framework import HostedCodeInterpreterTool | Python 코드 실행 |
HostedFileSearchTool | from agent_framework import HostedFileSearchTool | 벡터 저장소 검색 |
HostedWebSearchTool | from agent_framework import HostedWebSearchTool | Bing 웹 검색 |
HostedMCPTool | from agent_framework import HostedMCPTool | 서비스 관리 MCP |
MCPStreamableHTTPTool | from agent_framework import MCPStreamableHTTPTool | 클라이언트 관리 MCP |
완전한 예제
import asyncio
from typing import Annotated
from pydantic import BaseModel, Field
from agent_framework import (
HostedCodeInterpreterTool,
HostedWebSearchTool,
MCPStreamableHTTPTool,
)
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
def get_weather(
location: Annotated[str, Field(description="City name")],
) -> str:
"""Get weather for a location."""
return f"Weather in {location}: 72°F, sunny"
class AnalysisResult(BaseModel):
summary: str
key_findings: list[str]
confidence: float
async def main():
async with (
AzureCliCredential() as credential,
MCPStreamableHTTPTool(
name="Docs MCP",
url="https://learn.microsoft.com/api/mcp",
) as mcp_tool,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="ResearchAssistant",
instructions="You are a research assistant with multiple capabilities.",
tools=[
get_weather,
HostedCodeInterpreterTool(),
HostedWebSearchTool(name="Bing"),
mcp_tool,
],
)
thread = agent.get_new_thread()
# 비스트리밍
result = await agent.run(
"Search for Python best practices and summarize",
thread=thread,
)
print(f"Response: {result.text}")
# 스트리밍
print("\nStreaming: ", end="")
async for chunk in agent.run_stream("Continue with examples", thread=thread):
if chunk.text:
print(chunk.text, end="", flush=True)
print()
# 구조화된 출력
result = await agent.run(
"Analyze findings",
thread=thread,
response_format=AnalysisResult,
)
analysis = AnalysisResult.model_validate_json(result.text)
print(f"\nConfidence: {analysis.confidence}")
if __name__ == "__main__":
asyncio.run(main())
규칙
- 항상 비동기 컨텍스트 관리자 사용:
async with provider: - 함수를
tools=매개변수에 직접 전달 (자동으로 AIFunction으로 변환됨) - 함수 매개변수에는
Annotated[type, Field(description=...)]사용 - 다중 턴 대화에는
get_new_thread()사용 - 서비스 관리 MCP에는
HostedMCPTool, 클라이언트 관리 MCP에는MCPStreamableHTTPTool우선 사용
모범 사례
- 이 SDK는 비동기 우선(async-first)입니다. 코드 전체에
async def핸들러와async with를 사용하십시오. - 클라이언트와 비동기 자격 증명에는 항상 컨텍스트 관리자를 사용하십시오. 모든 클라이언트를
with Client(...) as client:(동기) 또는async with Client(...) as client:(비동기)로 래핑하십시오. azure.identity.aio의 비동기DefaultAzureCredential의 경우,async with credential:도 사용하여 토큰과 전송이 정리되도록 하십시오.
참조 파일
- references/tools.md: 호스팅 도구 패턴 상세
- references/mcp.md: MCP 통합 (호스팅 + 로컬)
- references/threads.md: 스레드 및 대화 관리
- references/advanced.md: OpenAPI, 인용, 구조화된 출력