Scenariusz: używanie zestawu Microsoft Entra SDK dla identyfikatora AgentID w języku Python

Utwórz bibliotekę klienta języka Python zintegrowaną z zestawem Microsoft Entra SDK for AgentID w celu uzyskania tokenów i wywołania podrzędnych interfejsów API. Następnie zintegruj tego klienta z aplikacjami Flask, FastAPI lub Django w celu obsługi uwierzytelnionych żądań.

Wymagania wstępne

• Konto Azure z aktywną subskrypcją. Utwórz konto bezpłatnie.
• Python (wersja 3.7 lub nowsza) z zainstalowanym pip na komputerze deweloperskim.
• Zestaw Microsoft Entra SDK dla AgentID wdrożony i uruchomiony w Twoim środowisku. Aby uzyskać instrukcje dotyczące instalacji, zobacz Przewodnik instalacji.
• Podrzędne interfejsy API skonfigurowane w zestawie SDK z podstawowymi adresami URL i wymaganymi zakresami.
• Odpowiednie uprawnienia w usłudze Microsoft Entra ID — Twoje konto musi mieć uprawnienia do rejestrowania aplikacji i udzielania uprawnień interfejsu API.

Konfiguracja

Przed utworzeniem biblioteki klienta zainstaluj wymaganą zależność do wykonywania żądań HTTP:

pip install requests

Implementacja biblioteki klienta

Utwórz klasę klienta wielokrotnego użytku, która opakowuje wywołania HTTP do zestawu Microsoft Entra SDK for AgentID. Ta klasa obsługuje przekazywanie tokenów, konfigurację żądań i obsługę błędów:

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

Integracja platformy Flask

Zintegruj bibliotekę klienta z aplikacją platformy Flask, wyodrębniając token przychodzący w funkcji pomocniczej i używając jej w programach obsługi tras w celu wywoływania podrzędnych interfejsów API:

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)

Integracja z interfejsem FastAPI

W przypadku aplikacji FastAPI użyj systemu wstrzykiwania zależności z zależnością Header, aby wyodrębnić i zweryfikować token autoryzacji przed przekazaniem go do obsługujących trasy:

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

Integracja z platformą Django

W przypadku aplikacji Django utwórz klasy widoków, które wyodrębnią token autoryzacji z nagłówków żądań i używają ich do wywoływania podrzędnych interfejsów API:

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

Zaawansowane: używanie żądań. Sesja

Aby uzyskać lepszą wydajność i odporność, użyj obiektu z logiką requests.Session ponawiania prób. Takie podejście umożliwia automatyczne ponawianie prób w przypadku przejściowych błędów i buforowania połączeń w celu zmniejszenia obciążenia:

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

Najlepsze rozwiązania

W przypadku korzystania z Microsoft Entra SDK dla identyfikatora AgentID w języku Python postępuj zgodnie z tymi praktykami, aby tworzyć niezawodne i łatwe do utrzymania aplikacje:

• Ponowne użycie wystąpienia klienta: utwórz pojedyncze SidecarClient wystąpienie i użyj go ponownie w całej aplikacji, zamiast tworzyć nowe wystąpienia na żądanie. Zwiększa to wydajność i użycie zasobów.
• Ustaw odpowiednie limity czasu: skonfiguruj limity czasu żądania na podstawie opóźnienia podrzędnego interfejsu API. Zapobiega to zawieszaniu się aplikacji przez czas nieokreślony, jeśli zestaw SDK lub usługa podrzędna działa wolno.
• Implementowanie obsługi błędów: Dodaj prawidłową obsługę błędów i logikę ponawiania prób, szczególnie w przypadku błędów przejściowych. Rozróżnianie błędów klienta (4xx) i błędów serwera (5xx) w celu określenia odpowiednich odpowiedzi.
• Użyj wskazówek dotyczących typów: dodaj wskazówki dotyczące typów do parametrów funkcji i zwracaj wartości, aby uzyskać lepszą przejrzystość kodu i przechwytywać błędy w czasie programowania.
• Włącz buforowanie połączeń: użyj requests.Session obiektu do ponownego użycia połączenia między żądaniami, co zmniejsza obciążenie i zwiększa przepływność wielu wywołań interfejsu API.

Inne przewodniki językowe

• Używanie z języka TypeScript
• Wzorce tożsamości zarządzanej

Dalsze kroki

Zacznij od scenariusza:

• Wywołaj podrzędne API
• Uzyskaj nagłówek autoryzacji
• Weryfikowanie nagłówka autoryzacji