Guía de Implementación: Bot de Conversación en Python con AI Foundry y Azure

Esta guía detalla cómo construir una arquitectura de bot escalable que conecta un backend en Python con un agente de Inteligencia Artificial (AI Foundry), lo despliega en Azure App Service y lo conecta de forma segura a Azure Bot Service utilizando el concepto de Service Principal.

Prerrequisitos

Un agente de AI Foundry configurado y que funcione correctamente en el playground de Azure AI Foundry.

Las credenciales para la conexión con el endpoint del AI Foundry.

Crear un recurso Bot Service para gestionar el bot y la exportación hacía los diferentes canales.

Una aplicación Service Principal, que bien podrías reutilizar la aplicación generada automáticamente por el Bot Service, la cual puedes visualizarla en la pestaña Configuración del Bot Service.

La aplicación del Service Principal debe tener añadida como una URL de redirección, la URL que se conecte al Bot Service.

La aplicación no requiere permisos específicos para conectarse directamente al AI Foundry. En su lugar, debe registrarse en el recurso de AI Foundry con los roles adecuados: Cognitive Services Contributor y Azure AI User. Estos permisos garantizan que la aplicación pueda interactuar correctamente con el servicio sin necesidad de configuraciones adicionales a nivel de la propia aplicación.

Arquitectura del Proyecto

La arquitectura se compone de tres partes principales:

1. Agente AI Foundry: (Externo) Provee la lógica conversacional a través de una API.

2. Backend Python (Flask/FastAPI): Aloja el endpoint /api/messages, gestiona la comunicación con el Agente AI y da formato a las respuestas. Desplegado en Azure App Service.

3. Azure Bot Service: Actúa como el puente, autenticándose con el backend y canalizando las conversaciones (Web Chat, Teams, etc.).

Configuración del Backend Python y AI Foundry

1.1. Preparación del Entorno Python

Asegúrate de tener un entorno virtual activo y los paquetes necesarios instalados.

# Core requirements
flask>=2.0.0  # Version con soporte async
python-dotenv>=0.19.0

# Azure Bot Framework
botbuilder-core==4.17.0
botbuilder-integration-aiohttp==4.17.0
botbuilder-schema==4.17.0
botframework-connector==4.17.0
botframework-streaming==4.17.0

# Azure
azure-ai-agents==1.1.0
azure-core==1.36.0
azure-identity==1.25.1

# Async support
aiohttp==3.13.2
aiosignal==1.4.0
async-timeout==5.0.1
aiohappyeyeballs==2.6.1

# Utils
blinker==1.9.0
certifi==2025.11.12
cffi==2.0.0
charset-normalizer==3.4.4
click==8.1.8
colorama==0.4.6
cryptography==46.0.3
attrs==25.4.0
Flask==3.0.0
frozenlist==1.8.0
gunicorn==23.0.0
idna==3.11
importlib_metadata==8.7.0
isodate==0.7.2
itsdangerous==2.2.0
Jinja2==3.1.6
jsonpickle==1.4.2
MarkupSafe==3.0.3
msal==1.34.0
msal-extensions==1.3.1
msrest==0.7.1
multidict==6.7.0
oauthlib==3.3.1
packaging==25.0
propcache==0.4.1
pycparser==2.23
PyJWT==2.10.1
python-dotenv==1.2.1
requests==2.32.5
requests-oauthlib==2.0.0
typing_extensions==4.15.0
urllib3==2.5.0
Werkzeug==3.0.1
yarl==1.22.0
zipp==3.23.0

Configuración del archivo app.py

En el archivo app.py, hay que configurar lo siguiente:

Leer configuración del Bot Service

APP_ID = os.environ.get("MicrosoftAppId", "")
APP_PASSWORD = os.environ.get("MicrosoftAppPassword", "")
APP_TENANT_ID = os.environ.get("MicrosoftAppTenantId", "")
SETTINGS = BotFrameworkAdapterSettings(APP_ID, APP_PASSWORD, APP_TENANT_ID)
ADAPTER = BotFrameworkAdapter(SETTINGS)
BOT = FoundryAgentBot()

Inicializar la aplicación de Flask

app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False

Crear el endpoint de Health Check

Permite verificar al momento del despliegue si el servicio fue desplegado correctamente.

@app.route("/", methods=["GET"])
def health_check():
    """
    Endpoint de health check que devuelve el estado del servicio.
    """
    return Response("Bot Service is Running and Healthy", status=200, mimetype="text/plain")

Crear el endpoint que se comunica con el BotService

El estándar de la ruta es api/messages, recibe una petición HTTP de tipo JSON y que se autentica mediante un token de autorización Bearer.

@app.route("/api/messages", methods=["POST"])
def messages():
    if "application/json" not in request.headers.get("Content-Type", ""):
        return Response("Unsupported Media Type", status=415)

    try:
        # Obtener el body y el header de autorización
        body = request.json
        auth_header = request.headers.get("Authorization", "")
        # Deserializar la actividad
        activity = Activity.deserialize(body)

        # Procesar la actividad de forma asíncrona
        async def process():
            await ADAPTER.process_activity(
                activity, 
                auth_header, 
                lambda context: BOT.on_turn(context)
            )

        # Ejecutar la función async (asyncio.run crea un nuevo event loop)
        asyncio.run(process())
        return Response(status=200)

    except Exception as e:
        logging.error(f"Error procesando la actividad: {e}")
        return Response(str(e), status=500, mimetype="text/plain")

Iniciar la aplicación Flask

if __name__ == "__main__":
    port = int(os.environ.get("PORT", 5000))
    host = os.environ.get("HOST", "0.0.0.0")
    app.run(host=host, port=port, debug=True)

Configuración del archivo foundry.py

Leer las configuraciones del AI Foundry

PROJECT_ENDPOINT = os.environ.get("AZURE_AI_FOUNDRY_PROJECT_ENDPOINT")
AGENT_ID = os.environ.get("AZURE_AI_FOUNDRY_AGENT_ID")

Inicializar la clase que gestiona la conexión al Foundry

class FoundryAgentBot(ActivityHandler):
    def __init__(self):
        # Autenticación: DefaultAzureCredential busca credenciales en el entorno
        self.credential = DefaultAzureCredential()
        # Cliente para interactuar con los servicios de Agentes de Foundry
        self.agent_client = AgentsClient(endpoint=PROJECT_ENDPOINT, credential=self.credential)
        # Diccionario para almacenar el Thread ID de Foundry por usuario/conversación
        self.user_threads = {}

Añadir el método para gestionar la actividad de los mensajes

Primero hay que definir la función junto con los parámetros que requiere.

async def on_message_activity(self, turn_context: TurnContext):
    """Maneja el mensaje de entrada del usuario."""

Obtener el contexto de Foundry y crear un hilo de conversación

# Obtener el ID de la conversación
conversation_id = turn_context.activity.conversation.id
# Generar un identificador para el hilo de la conversación
thread_id = self.user_threads.get(conversation_id)

# Crear un nuevo hilo en caso no exista uno
if not thread_id:
    new_thread = self.agent_client.threads.create()
    thread_id = new_thread.id
    self.user_threads[conversation_id] = thread_id
    await turn_context.send_activity(
        "¡Hola! Soy un agente impulsado por Azure AI Foundry. ¿En qué puedo ayudarte?"
    )

# Enviar el mensaje del usuario como actividad al agente de Foundry
user_message = turn_context.activity.text
await turn_context.send_activity(f"Enviando el mensaje del usuario hacía AI Foundry...** '{user_message}'")

# Crear el cliente del agente con el contexto, el rol 'Usuario' y el identificador del Thread.
try:
    self.agent_client.messages.create(
        thread_id=thread_id,
        role="user",
        content=user_message
    )
except AttributeError as e:
    raise Exception(f"Error al crear mensaje: {e}. Revisar sintaxis de create_message.")

Implementar un bucle de polling para la conexión al momento de obtener los mensajes

 # Implementar el bucle de POLLING (Sondeo) manual
while run.status not in ["completed", "failed", "cancelled", "expired"]:
    # Esperar 1 segundo antes de verificar el estado nuevamente
    await asyncio.sleep(1) 

    # Obtener el estado actualizado del run usando el método 'get' del objeto 'runs'
    run = self.agent_client.runs.get(
        thread_id=thread_id,
        run_id=run.id
    )

Obtener la respuesta del agente

if run.status == "completed":
    messages_result = self.agent_client.messages.list( 
        thread_id=thread_id, 
        order="desc", 
        limit=1
    )
    messages_list = list(messages_result) 
    # Obtener el contenido del mensaje
    if messages_list and messages_list[0] and messages_list[0].content:
        result_message = messages_list[0].content[0]
        result_response = result_message['text']['value']
        agent_response = result_response
    else:
        agent_response = "No se recibió respuesta del agente."
else:
    agent_response = f"El agente no pudo completar la tarea. Estado: {run.status}. Error: {run.last_error}"

Responder al usuario con el mensaje que retorna el agente de AI Foundry

await turn_context.send_activity(agent_response)

Una vez desplegado el proyecto en un APP Service, se debe configurar las variables de entorno, una configuración similar a la siguiente.

Y finalmente en el Bot Service accediendo al apartado de Canales, se puede probar utilizando el canal web que es la forma más rápida.

Bonus

El siguiente código permite manejar el evento cuando el Bot se despliegue en un entorno como Microsoft Teams mediante los canales del Bot Service.

async def on_members_added_activity(self, members_added: list[ChannelAccount], turn_context: TurnContext):
    "Maneja cuando el bot es añadido a una conversación (útil en Teams)."
    for member in members_added:
        if member.id != turn_context.activity.recipient.id:
            await turn_context.send_activity(
                Activity(
                    type=ActivityTypes.message, 
                    text="¡Hola! Soy un bot que usa un Agente de Azure AI Foundry. Puedes empezar a chatear conmigo."
                )
            )

Además si necesitas probar generando tu propio token puedes hacerlo mediante este pequeño script de Python.

# Importar librerias
from azure.identity import DefaultAzureCredential
from datetime import datetime

# Inicializar credencial, utilizará los valores dependiendo del entorno que tengas, sea proporcionando
# credenciales de Service Principal o de una identidad gestionada del sistema
credential = DefaultAzureCredential()

# Scope para Microsoft Graph (puedes cambiarlo según tu necesidad)
scope = "https://graph.microsoft.com/.default"

# Obtener token
token = credential.get_token(scope)
expires_readable = datetime.utcfromtimestamp(token.expires_on).strftime('%Y-%m-%d %H:%M:%S')
print("Access Token:", token.token)
print("Expires On:", token.expires_on)
print("Expires On expires_readable:", expires_readable)

Por último si deseas probar el chat mediante un endpoint HTTP, puedes hacerlo mediante el siguiente código que añade una ruta que la puedes probar desde un cliente como Postman, Insomnia o un Frontend.

@app.route("/api/chat", methods=["POST"])
def chat():
    """
    Endpoint HTTP simple para probar el bot con peticiones REST estándar.
    Acepta un JSON con un campo 'message' y devuelve la respuesta del bot.
    """
    if not request.is_json:
        return jsonify({"error": "Content type must be application/json"}), 415

    try:
        data = request.get_json()
        user_message = data.get('message', '')

        if not user_message:
            return jsonify({"error": "Field 'message' is required"}), 400

        # Crear una actividad de mensaje simple con todos los campos requeridos
        from botbuilder.schema import ConversationAccount
        from botbuilder.core import TurnContext, BotAdapter

        # Crear la actividad con los campos mínimos requeridos
        activity = Activity(
            type="message",
            text=user_message,
            channel_id="http",
            from_property={"id": "http-user", "name": "HTTP User", "role": "user"},
            recipient={"id": "bot", "role": "bot"},
            conversation=ConversationAccount(
                id="http-conversation-123",  # ID de conversación fijo para pruebas
                name="Test Conversation"
            ),
            service_url="http://localhost:5000", # Reemplazar con la URL del servicio si es necesario cuando se use en producción
            channel_data={},
            reply_to_id=""
        )

        # Variable para almacenar la respuesta
        response_text = "No se recibió respuesta del bot"

        # Crear un adaptador simple para el contexto
        class SimpleAdapter(BotAdapter):
            def __init__(self):
                super().__init__()
                self.sent_messages = []

            async def send_activities(self, context, activities):
                responses = []
                for activity in activities:
                    self.sent_messages.append(activity)
                    responses.append({'id': '123'})
                return responses

            async def delete_activity(self, context, reference):
                """Delete an existing activity."""
                # Implementación básica para cumplir con la interfaz
                return

            async def update_activity(self, context, activity):
                """Update an existing activity."""
                # Implementación básica para cumplir con la interfaz
                return {'id': '123'}

        # Crear el contexto de turno
        adapter = SimpleAdapter()
        context = TurnContext(adapter, activity)
        context.turn_state['BotCallbackHandler'] = None

        # Procesar la actividad de forma síncrona
        import asyncio

        async def process():
            await BOT.on_turn(context)

            # Obtener la respuesta del bot
            if hasattr(adapter, 'sent_messages') and adapter.sent_messages:
                last_message = adapter.sent_messages[-1]
                if hasattr(last_message, 'text'):
                    return last_message.text
                elif isinstance(last_message, dict) and 'text' in last_message:
                    return last_message['text']
                return str(last_message)
            return "No se recibió respuesta del bot"

        # Ejecutar la corutina y esperar el resultado
        response_text = asyncio.run(process())

        # Devolver la respuesta en formato JSON
        return jsonify({"response": response_text})

    except Exception as e:
        logging.error(f"Error en /api/chat: {str(e)}", exc_info=True)
        return jsonify({"error": str(e)}), 500