SamuelChien/mega-skills-collection

Agent Framework Azure Hosted Agents

Construye agentes persistentes en Azure AI Foundry usando el SDK de Python del Microsoft Agent Framework.

Arquitectura

User Query → AzureAIAgentsProvider → Azure AI Agent Service (Persistent)
                    ↓
              Agent.run() / Agent.run_stream()
                    ↓
              Tools: Functions | Hosted (Code/Search/Web) | MCP
                    ↓
              AgentThread (persistencia de conversación)

Instalación

# Framework completo (recomendado)
pip install agent-framework --pre

# O solo el paquete específico de Azure
pip install agent-framework-azure-ai --pre

Variables de Entorno

export AZURE_AI_PROJECT_ENDPOINT="https://<project>.services.ai.azure.com/api/projects/<project-id>"  # Requerido para todos los métodos de autenticación
export AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini"  # Requerido para todos los métodos de autenticación
export BING_CONNECTION_ID="your-bing-connection-id"  # Para búsqueda web
export AZURE_TOKEN_CREDENTIALS=prod # Requerido solo si se usa DefaultAzureCredential en producción

Autenticación y Ciclo de Vida

🔑 Dos reglas aplican a cada fragmento de código a continuación:

1. Preferir DefaultAzureCredential. Funciona localmente (Azure CLI / VS Code / Developer CLI) y en Azure (managed identity, workload identity) sin cambios de código. Evita cadenas de conexión, claves de cuenta/API — evitan la auditoría y rotación de Entra.
   • Desarrollo local: DefaultAzureCredential funciona tal cual.
   • Producción: establece AZURE_TOKEN_CREDENTIALS=prod (o AZURE_TOKEN_CREDENTIALS=<credential_específico>) para restringir la cadena de credenciales a aquellas seguras para producción.
2. Envuelve cada cliente en un context manager para que los transportes HTTP, sockets y cachés de tokens se liberen de manera determinista:
   • Sincrónico: with <Cliente>(...) as client:
   • Asincrónico: async with <Cliente>(...) as client: y async with DefaultAzureCredential() as credential: (de azure.identity.aio)

Los fragmentos pueden abreviar esta configuración, pero el código en producción debe seguir siempre ambas reglas.

from azure.identity.aio import AzureCliCredential, DefaultAzureCredential, ManagedIdentityCredential

# Desarrollo
credential = AzureCliCredential()

# Producción
# Desarrollo local: DefaultAzureCredential. Producción: establecer AZURE_TOKEN_CREDENTIALS=prod o AZURE_TOKEN_CREDENTIALS=<credential_específico>
credential = DefaultAzureCredential(require_envvar=True)
# O usar una credencial específica directamente en producción:
# Ver https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

Flujo de Trabajo Principal

Agente Básico

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="Eres un asistente útil.",
        )
        
        result = await agent.run("¡Hola!")
        print(result.text)

asyncio.run(main())

Agente con Herramientas de Función

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="Nombre de la ciudad para obtener el clima")],
) -> str:
    """Obtiene el clima actual para una ubicación."""
    return f"Clima en {location}: 72°F, soleado"

def get_current_time() -> str:
    """Obtiene la hora UTC actual."""
    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="Ayudas con consultas sobre el clima y la hora.",
            tools=[get_weather, get_current_time],  # Pasa funciones directamente
        )
        
        result = await agent.run("¿Cómo está el clima en Seattle?")
        print(result.text)

Agente con Herramientas Alojadas

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="Puedes ejecutar código, buscar archivos y buscar en la web.",
            tools=[
                HostedCodeInterpreterTool(),
                HostedWebSearchTool(name="Bing"),
            ],
        )
        
        result = await agent.run("Calcula el factorial de 20 en Python")
        print(result.text)

Respuestas en Streaming

async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="StreamingAgent",
            instructions="Eres un asistente útil.",
        )
        
        print("Agente: ", end="", flush=True)
        async for chunk in agent.run_stream("Cuéntame una historia corta"):
            if chunk.text:
                print(chunk.text, end="", flush=True)
        print()

Hilos de Conversación

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="Eres un asistente útil.",
            tools=[get_weather],
        )
        
        # Crear hilo para persistencia de la conversación
        thread = agent.get_new_thread()
        
        # Primer turno
        result1 = await agent.run("¿Cómo está el clima en Seattle?", thread=thread)
        print(f"Agente: {result1.text}")
        
        # Segundo turno - se mantiene el contexto
        result2 = await agent.run("¿Y en Portland?", thread=thread)
        print(f"Agente: {result2.text}")
        
        # Guardar ID del hilo para reanudar después
        print(f"ID de conversación: {thread.conversation_id}")

Salidas Estructuradas

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="Proporciona información del clima en formato estructurado.",
            response_format=WeatherResponse,
        )
        
        result = await agent.run("¿Clima en Seattle?")
        weather = WeatherResponse.model_validate_json(result.text)
        print(f"{weather.location}: {weather.temperature}°{weather.unit}")

Métodos del Proveedor

Método	Descripción
create_agent()	Crea un nuevo agente en el servicio Azure AI
get_agent(agent_id)	Recupera un agente existente por ID
as_agent(sdk_agent)	Envuelve un objeto SDK Agent (sin llamada HTTP)

Referencia Rápida de Herramientas Alojadas

Herramienta	Importación	Propósito
HostedCodeInterpreterTool	from agent_framework import HostedCodeInterpreterTool	Ejecuta código Python
HostedFileSearchTool	from agent_framework import HostedFileSearchTool	Busca en vector stores
HostedWebSearchTool	from agent_framework import HostedWebSearchTool	Búsqueda web con Bing
HostedMCPTool	from agent_framework import HostedMCPTool	MCP gestionado por el servicio
MCPStreamableHTTPTool	from agent_framework import MCPStreamableHTTPTool	MCP gestionado por el cliente

Ejemplo Completo

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="Nombre de la ciudad")],
) -> str:
    """Obtiene el clima para una ubicación."""
    return f"Clima en {location}: 72°F, soleado"


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="Eres un asistente de investigación con múltiples capacidades.",
            tools=[
                get_weather,
                HostedCodeInterpreterTool(),
                HostedWebSearchTool(name="Bing"),
                mcp_tool,
            ],
        )
        
        thread = agent.get_new_thread()
        
        # Sin streaming
        result = await agent.run(
            "Busca las mejores prácticas de Python y haz un resumen",
            thread=thread,
        )
        print(f"Respuesta: {result.text}")
        
        # Streaming
        print("\nStreaming: ", end="")
        async for chunk in agent.run_stream("Continúa con ejemplos", thread=thread):
            if chunk.text:
                print(chunk.text, end="", flush=True)
        print()
        
        # Salida estructurada
        result = await agent.run(
            "Analiza los hallazgos",
            thread=thread,
            response_format=AnalysisResult,
        )
        analysis = AnalysisResult.model_validate_json(result.text)
        print(f"\nConfianza: {analysis.confidence}")


if __name__ == "__main__":
    asyncio.run(main())

Convenciones

• Usa siempre context managers asíncronos: async with provider:
• Pasa funciones directamente al parámetro tools= (se convierten automáticamente a AIFunction)
• Usa Annotated[type, Field(description=...)] para parámetros de funciones
• Usa get_new_thread() para conversaciones de múltiples turnos
• Prefiere HostedMCPTool para MCP gestionado por el servicio, MCPStreamableHTTPTool para MCP gestionado por el cliente

Mejores Prácticas

1. Este SDK es asíncrono primero — usa async def handlers y async with en todo momento.
2. Usa siempre context managers para clientes y credenciales asíncronas. Envuelve cada cliente en with Client(...) as client: (sincrónico) o async with Client(...) as client: (asíncrono). Para DefaultAzureCredential asíncrono de azure.identity.aio, también usa async with credential: para que los tokens y transportes se limpien.

Archivos de Referencia

• references/tools.md: Patrones detallados de herramientas alojadas
• references/mcp.md: Integración MCP (alojada + local)
• references/threads.md: Gestión de hilos y conversaciones
• references/advanced.md: OpenAPI, citas, salidas estructuradas