Szenario: Verwenden des Microsoft Entra SDK für AgentID aus Python

Erstellen Sie eine Python-Clientbibliothek, die in das Microsoft Entra SDK für AgentID integriert ist, um Token abzurufen und nachgeschaltete APIs aufzurufen. Integrieren Sie diesen Client dann in Flask-, FastAPI- oder Django-Anwendungen, um authentifizierte Anforderungen zu verarbeiten.

Voraussetzungen

Ein Azure-Konto mit einem aktiven Abonnement. Kostenlos ein Konto erstellen.
Python (Version 3.7 oder höher) mit Pip auf Ihrem Entwicklungscomputer installiert.
Das Microsoft Entra SDK für AgentID wird in Ihrer Umgebung bereitgestellt und ausgeführt. Anweisungen zur Einrichtung finden Sie im Installationshandbuch .
Downstream-APIs , die im SDK mit Basis-URLs und erforderlichen Bereichen konfiguriert sind.
Geeignete Berechtigungen in microsoft Entra ID – Ihr Konto muss über Berechtigungen zum Registrieren von Anwendungen verfügen und API-Berechtigungen erteilen.

Konfiguration

Bevor Sie Ihre Clientbibliothek erstellen, installieren Sie die erforderliche Abhängigkeit zum Erstellen von HTTP-Anforderungen:

pip install requests

Clientbibliotheksimplementierung

Erstellen Sie eine wiederverwendbare Clientklasse, die HTTP-Aufrufe an das Microsoft Entra SDK für AgentID umschließt. Diese Klasse behandelt die Tokenweiterleitung, Anforderungskonfiguration und Fehlerbehandlung:

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

Flask-Integration

Integrieren Sie die Clientbibliothek in eine Flask-Anwendung, indem Sie das eingehende Token in einer Hilfsfunktion extrahieren und in Routinghandlern verwenden, um nachgeschaltete APIs aufzurufen:

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)

FastAPI-Integration

Verwenden Sie für FastAPI-Anwendungen das Abhängigkeitseinfügungssystem mit der Header Abhängigkeit, um das Autorisierungstoken zu extrahieren und zu überprüfen, bevor Sie es an Routinghandler übergeben:

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

Django-Integration

Erstellen Sie für Django-Anwendungen Ansichtsklassen, die das Autorisierungstoken aus Anforderungsheadern extrahieren, und verwenden Sie sie, um nachgeschaltete APIs aufzurufen:

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

Fortgeschritten: Nutzung von requests.Session

Um die Leistung und Resilienz zu verbessern, verwenden Sie ein requests.Session Objekt mit Wiederholungslogik. Dieser Ansatz ermöglicht automatische Wiederholungen für vorübergehende Fehler und Verbindungspooling, um den Aufwand zu reduzieren:

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

Bewährte Methoden

Wenn Sie das Microsoft Entra SDK für AgentID aus Python verwenden, befolgen Sie die folgenden Methoden, um zuverlässige und wartungsfähige Anwendungen zu erstellen:

Clientinstanz wiederverwenden: Erstellen Sie eine einzelne SidecarClient Instanz, und verwenden Sie sie in der gesamten Anwendung wieder, anstatt neue Instanzen pro Anforderung zu erstellen. Dadurch wird die Leistung und die Ressourcennutzung verbessert.
Legen Sie geeignete Timeouts fest: Konfigurieren Sie Anforderungstimeouts basierend auf der nachgeschalteten API-Latenz. Dadurch wird verhindert, dass Ihre Anwendung unbegrenzt hängen bleibt, wenn der SDK- oder downstream-Dienst langsam ist.
Implementieren der Fehlerbehandlung: Fügen Sie eine ordnungsgemäße Fehlerbehandlungs- und Wiederholungslogik hinzu, insbesondere für vorübergehende Fehler. Unterscheiden Sie zwischen Clientfehlern (4xx) und Serverfehlern (5xx), um geeignete Antworten zu ermitteln.
Verwenden Sie Typenhinweise: Hinzufügen von Typenhinweisen zu Funktionsparametern und Rückgabewerten zur besseren Codeklarheit und zum Abfangen von Fehlern während der Entwicklungsphase.
Aktivieren von Verbindungspooling: Verwenden Sie ein requests.Session Objekt für die Wiederverwendung zwischen Anforderungen, wodurch der Aufwand reduziert und der Durchsatz für mehrere API-Aufrufe verbessert wird.

Andere Sprachhandbücher

Nächste Schritte

Beginnen Sie mit einem Szenario:

Feedback

War diese Seite hilfreich?

Last updated on 2025-11-14