Scenario: Använda Microsoft Entra SDK för AgentID från Python

Skapa ett Python-klientbibliotek som integreras med Microsoft Entra SDK för AgentID för att hämta token och anropa underordnade API:er. Integrera sedan den här klienten i Flask-, FastAPI- eller Django-program för att hantera autentiserade begäranden.

Förutsättningar

• Ett Azure-konto med en aktiv prenumeration. Skapa ett konto kostnadsfritt.
• Python (version 3.7 eller senare) med pip installerat på utvecklingsdatorn.
• Microsoft Entra SDK för AgentID distribueras och körs i din miljö. Se Installationsguide för installationsinstruktioner.
• Underordnade API:er som konfigurerats i SDK med bas-URL:er och nödvändiga omfång.
• Lämpliga behörigheter i Microsoft Entra-ID – Ditt konto måste ha behörighet att registrera program och bevilja API-behörigheter.

Inställningar

Innan du skapar klientbiblioteket installerar du det beroende som krävs för att göra HTTP-begäranden:

pip install requests

Implementering av klientbibliotek

Skapa en återanvändbar klientklass som omsluter HTTP-anrop till Microsoft Entra SDK för AgentID. Den här klassen hanterar vidarebefordran av token, konfiguration av begäranden och felhantering:

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

Integrera klientbiblioteket i ett Flask-program genom att extrahera den inkommande token i en hjälpfunktion och använda den i routningshanterare för att anropa underordnade API:er:

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

För FastAPI-program använder du beroendeinmatningssystemet med Header beroendet för att extrahera och verifiera auktoriseringstoken innan du skickar den till routningshanterare:

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

För Django-program skapar du vyklasser som extraherar auktoriseringstoken från begärandehuvuden och använder den för att anropa underordnade API:er:

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

Avancerat: att använda requests.Session

För bättre prestanda och motståndskraft använder du ett requests.Session objekt med logik för återförsök. Den här metoden möjliggör automatiska återförsök för tillfälliga fel och anslutningspooler för att minska kostnaderna:

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

Metodtips

När du använder Microsoft Entra SDK för AgentID från Python följer du dessa metoder för att skapa tillförlitliga och underhållsbara program:

• Återanvänd klientinstans: Skapa en enskild SidecarClient instans och återanvänd den i hela programmet i stället för att skapa nya instanser per begäran. Detta förbättrar prestanda och resursanvändning.
• Ange lämpliga tidsgränser: Konfigurera tidsgränser för begäran baserat på din nedströms-API-svarstid. Detta förhindrar att ditt program hänger sig på obestämd tid om SDK:t eller nedströmstjänsten är långsam.
• Implementera felhantering: Lägg till korrekt felhantering och omförsökslogik, särskilt för tillfälliga fel. Skilja mellan klientfel (4xx) och serverfel (5xx) för att fastställa lämpliga svar.
• Använd typtips: Lägg till typtips till funktionsparametrar och returnera värden för bättre kod klarhet och för att fånga fel vid utvecklingstidpunkt.
• Aktivera anslutningspooler: Använd ett requests.Session objekt för återanvändning av anslutningar mellan begäranden, vilket minskar omkostnaderna och förbättrar dataflödet för flera API-anrop.

Andra språkguider

• Använda från TypeScript
• Mönster för hanterad identitet

Nästa steg

Börja med ett scenario:

• Anropa nedströms-API
• Hämta autentiseringshuvud
• Verifiera auktoriseringshuvud