Scenario: uso dell'SDK di Microsoft Entra per l'ID agente da Python

Creare una libreria client Python che si integra con Microsoft Entra SDK per l'ID agente per ottenere i token e chiamare le API downstream. Integrare quindi questo client in applicazioni Flask, FastAPI o Django per gestire le richieste autenticate.

Prerequisiti

  • Un account Azure con una sottoscrizione attiva. Creare un account gratuito.
  • Python (versione 3.7 o successiva) con pip installato nel computer di sviluppo.
  • Microsoft Entra SDK per AgentID distribuito ed eseguito nel tuo ambiente. Per istruzioni sull'installazione, vedere Guida all'installazione .
  • API downstream configurate nell'SDK con URL di base e ambiti obbligatori.
  • Autorizzazioni Appropriate in Microsoft Entra ID: Il tuo account deve disporre delle autorizzazioni per registrare applicazioni e concedere autorizzazioni API.

Configurazione

Prima di creare la libreria client, installare la dipendenza necessaria per l'esecuzione di richieste HTTP:

pip install requests

Implementazione della libreria client

Creare una classe client riutilizzabile che esegue il wrapping delle chiamate HTTP al Microsoft Entra SDK per AgentID. Questa classe gestisce l'inoltro dei token, la configurazione delle richieste e la gestione degli errori:

# 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')

Integrazione di Flask

Integrare la libreria client in un'applicazione Flask estraendo il token in ingresso in una funzione helper e usandolo nei gestori di route per chiamare le API downstream:

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)

Integrazione fastAPI

Per le applicazioni FastAPI, usare il sistema di inserimento delle dipendenze con la Header dipendenza per estrarre e convalidare il token di autorizzazione prima di passarlo ai gestori di route:

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))

Integrazione di Django

Per le applicazioni Django, creare classi di visualizzazione che estraggono il token di autorizzazione dalle intestazioni della richiesta e usarle per chiamare le API downstream:

# 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)

Avanzate: uso delle richieste. Sessione

Per migliorare le prestazioni e la resilienza, usare un requests.Session oggetto con logica di ripetizione dei tentativi. Questo approccio consente di ritentare automaticamente gli errori temporanei e il pool di connessioni per ridurre il sovraccarico:

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

Procedure consigliate

Quando si usa Microsoft Entra SDK per l'ID agente da Python, seguire queste procedure per creare applicazioni affidabili e gestibili:

  • Riutilizzare l'istanza client: creare una singola SidecarClient istanza e riutilizzarla in tutta l'applicazione invece di creare nuove istanze per ogni richiesta. Ciò migliora le prestazioni e l'utilizzo delle risorse.
  • Impostare timeout appropriati: configurare i timeout delle richieste in base alla latenza dell'API downstream. Ciò impedisce all'applicazione di bloccarsi a tempo indefinito se l'SDK o il servizio downstream è lento.
  • Implementare la gestione degli errori: aggiungere una corretta gestione degli errori e logica di ripetizione dei tentativi, in particolare per gli errori temporanei. Distinguere tra errori client (4xx) ed errori del server (5xx) per determinare le risposte appropriate.
  • Usare hint di tipo: aggiungere hint di tipo ai parametri di funzione e restituire valori per una migliore chiarezza del codice e per rilevare gli errori in fase di sviluppo.
  • Abilita pool di connessioni: usare un requests.Session oggetto per il riutilizzo della connessione tra le richieste, riducendo il sovraccarico e migliorando la velocità effettiva per più chiamate API.

Contenuti correlati

  • Last updated on