Escenario: Uso del SDK de Microsoft Entra para AgentID desde Python

Cree una biblioteca cliente de Python que se integre con el SDK de Microsoft Entra para AgentID para obtener tokens y llamar a las API de bajada. A continuación, integre este cliente en aplicaciones flask, FastAPI o Django para controlar las solicitudes autenticadas.

Prerrequisitos

• Una cuenta de Azure con una suscripción activa. Cree una cuenta gratuita.
• Python (versión 3.7 o posterior) con pip instalado en la máquina de desarrollo.
• Microsoft Entra SDK para AgentID implementado y en ejecución en su entorno. Consulte la Guía de instalación para obtener instrucciones de configuración.
• APIs de cliente configuradas en el SDK con URLs base y ámbitos necesarios.
• Permisos adecuados en el identificador de Microsoft Entra : la cuenta debe tener permisos para registrar aplicaciones y conceder permisos de API.

Configuración

Antes de crear la biblioteca cliente, instale la dependencia necesaria para realizar solicitudes HTTP:

pip install requests

Implementación de la biblioteca cliente

Cree una clase de cliente reutilizable que encapsula las llamadas HTTP al SDK de Microsoft Entra para AgentID. Esta clase controla el reenvío de tokens, la configuración de solicitudes y el control de errores:

# sidecar_client.py
import os
import json
import requests
from typing import Dict, Any, Optional, List
from urllib.parse import urlencode

class SidecarClient:
    """Client for interacting with the Microsoft Entra SDK for AgentID."""
    
    def __init__(self, base_url: Optional[str] = None, timeout: int = 10):
        self.base_url = base_url or os.getenv('SIDECAR_URL', 'http://localhost:5000')
        self.timeout = timeout
    
    def get_authorization_header(
        self,
        incoming_token: str,
        service_name: str,
        scopes: Optional[List[str]] = None,
        tenant: Optional[str] = None,
        agent_identity: Optional[str] = None,
        agent_username: Optional[str] = None
    ) -> str:
        """Get authorization header from the SDK."""
        params = {}
        
        if scopes:
            params['optionsOverride.Scopes'] = scopes
        
        if tenant:
            params['optionsOverride.AcquireTokenOptions.Tenant'] = tenant
        
        if agent_identity:
            params['AgentIdentity'] = agent_identity
            if agent_username:
                params['AgentUsername'] = agent_username
        
        response = requests.get(
            f"{self.base_url}/AuthorizationHeader/{service_name}",
            params=params,
            headers={'Authorization': incoming_token},
            timeout=self.timeout
        )
        
        response.raise_for_status()
        data = response.json()
        return data['authorizationHeader']
    
    def call_downstream_api(
        self,
        incoming_token: str,
        service_name: str,
        relative_path: str,
        method: str = 'GET',
        body: Optional[Dict[str, Any]] = None,
        scopes: Optional[List[str]] = None
    ) -> Any:
        """Call downstream API via the SDK."""
        params = {'optionsOverride.RelativePath': relative_path}
        
        if method != 'GET':
            params['optionsOverride.HttpMethod'] = method
        
        if scopes:
            params['optionsOverride.Scopes'] = scopes
        
        headers = {'Authorization': incoming_token}
        json_body = None
        
        if body:
            headers['Content-Type'] = 'application/json'
            json_body = body
        
        response = requests.request(
            method,
            f"{self.base_url}/DownstreamApi/{service_name}",
            params=params,
            headers=headers,
            json=json_body,
            timeout=self.timeout
        )
        
        response.raise_for_status()
        data = response.json()
        
        if data['statusCode'] >= 400:
            raise Exception(f"API error {data['statusCode']}: {data['content']}")
        
        return json.loads(data['content'])

# Usage
sidecar = SidecarClient(base_url='http://localhost:5000')

# Get authorization header
auth_header = sidecar.get_authorization_header(token, 'Graph')

# Call API
profile = sidecar.call_downstream_api(token, 'Graph', 'me')

Integración de Flask

Integre la biblioteca cliente en una aplicación de Flask mediante la extracción del token entrante en una función auxiliar y su uso en controladores de rutas para llamar a las API de bajada:

from flask import Flask, request, jsonify
from sidecar_client import SidecarClient

app = Flask(__name__)
sidecar = SidecarClient()

def get_token():
    """Extract token from request."""
    token = request.headers.get('Authorization')
    if not token:
        raise ValueError('No authorization token provided')
    return token

@app.route('/api/profile')
def profile():
    try:
        token = get_token()
        profile_data = sidecar.call_downstream_api(
            token,
            'Graph',
            'me'
        )
        return jsonify(profile_data)
    except ValueError as e:
        return jsonify({'error': str(e)}), 401
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/messages')
def messages():
    try:
        token = get_token()
        messages_data = sidecar.call_downstream_api(
            token,
            'Graph',
            'me/messages?$top=10'
        )
        return jsonify(messages_data)
    except ValueError as e:
        return jsonify({'error': str(e)}), 401
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/messages/send', methods=['POST'])
def send_message():
    try:
        token = get_token()
        message = request.json
        
        result = sidecar.call_downstream_api(
            token,
            'Graph',
            'me/sendMail',
            method='POST',
            body={'message': message}
        )
        
        return jsonify({'success': True, 'result': result})
    except ValueError as e:
        return jsonify({'error': str(e)}), 401
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

Integración de FastAPI

Para las aplicaciones de FastAPI, use el sistema de inserción de dependencias con la Header dependencia para extraer y validar el token de autorización antes de pasarlo a los controladores de ruta:

from fastapi import FastAPI, Header, HTTPException
from sidecar_client import SidecarClient
from typing import Optional

app = FastAPI()
sidecar = SidecarClient()

async def get_token(authorization: Optional[str] = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="No authorization token")
    return authorization

@app.get("/api/profile")
async def get_profile(token: str = Depends(get_token)):
    try:
        return sidecar.call_downstream_api(token, 'Graph', 'me')
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/messages")
async def get_messages(token: str = Depends(get_token)):
    try:
        return sidecar.call_downstream_api(
            token,
            'Graph',
            'me/messages?$top=10'
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Integración de Django

En el caso de las aplicaciones de Django, cree clases de vista que extraigan el token de autorización de los encabezados de solicitud y úsela para llamar a las API de bajada:

# views.py
from django.http import JsonResponse
from django.views import View
from sidecar_client import SidecarClient

sidecar = SidecarClient()

class ProfileView(View):
    def get(self, request):
        token = request.META.get('HTTP_AUTHORIZATION')
        if not token:
            return JsonResponse({'error': 'No authorization token'}, status=401)
        
        try:
            profile = sidecar.call_downstream_api(token, 'Graph', 'me')
            return JsonResponse(profile)
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=500)

class MessagesView(View):
    def get(self, request):
        token = request.META.get('HTTP_AUTHORIZATION')
        if not token:
            return JsonResponse({'error': 'No authorization token'}, status=401)
        
        try:
            messages = sidecar.call_downstream_api(
                token,
                'Graph',
                'me/messages?$top=10'
            )
            return JsonResponse(messages)
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=500)

Avanzado: mediante solicitudes. Sesión

Para mejorar el rendimiento y la resistencia, use un requests.Session objeto con lógica de reintento. Este enfoque permite reintentos automáticos para errores transitorios y agrupación de conexiones para reducir la sobrecarga:

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

class SidecarClient:
    def __init__(self, base_url: Optional[str] = None):
        self.base_url = base_url or os.getenv('SIDECAR_URL', 'http://localhost:5000')
        
        # Configure session with retry logic
        self.session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=0.3,
            status_forcelist=[500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount('http://', adapter)
        self.session.mount('https://', adapter)
    
    def call_downstream_api(self, token, service_name, relative_path, **kwargs):
        # Use self.session instead of requests
        response = self.session.get(...)
        return response

procedimientos recomendados

Al usar el SDK de Microsoft Entra para AgentID desde Python, siga estos procedimientos para crear aplicaciones confiables y fáciles de mantener:

• Reutilización de la instancia de cliente: cree una sola SidecarClient instancia y reutilícela en toda la aplicación en lugar de crear nuevas instancias por solicitud. Esto mejora el rendimiento y el uso de recursos.
• Tiempos de espera apropiados: Configure los tiempos de espera de solicitud en función de la latencia de la API descendente. Esto impide que la aplicación se bloquee indefinidamente si el SDK o el servicio de bajada son lentos.
• Implementar manejo de errores: agregue un manejo de errores adecuado y una lógica de reintento, especialmente para fallos transitorios. Distinguir entre los errores de cliente (4xx) y los errores del servidor (5xx) para determinar las respuestas adecuadas.
• Usar sugerencias de tipo: agregue sugerencias de tipo a parámetros de función y valores devueltos para mejorar la claridad del código y detectar errores en tiempo de desarrollo.
• Habilitar la agrupación de conexiones: use un requests.Session objeto para reutilizar la conexión entre solicitudes, lo que reduce la sobrecarga y mejora el rendimiento de varias llamadas API.

Otras guías de idioma

• Uso desde TypeScript
• Patrones de identidad administrada

Pasos siguientes

Comience con un escenario:

• Llamar a la API descendente
• Obtener encabezado de autorización
• Validar encabezado de autorización