Skenario: Menggunakan Microsoft Entra SDK untuk AgentID dari Python

Buat pustaka klien Python yang terintegrasi dengan Microsoft Entra SDK for AgentID untuk mendapatkan token dan memanggil API hilir. Kemudian integrasikan klien ini ke dalam aplikasi Flask, FastAPI, atau Django untuk menangani permintaan terautentikasi.

Prasyarat

Sebuah akun Azure dengan langganan aktif. Buat akun secara gratis.
Python (versi 3.7 atau yang lebih baru) dengan pip yang diinstal pada komputer pengembangan Anda.
Microsoft Entra SDK for AgentID disebarkan dan berjalan di lingkungan Anda. Lihat Panduan Penginstalan untuk instruksi penyiapan.
API hilir yang dikonfigurasi di SDK dengan URL dasar dan cakupan yang diperlukan.
Izin yang sesuai di ID Microsoft Entra - Akun Anda harus memiliki izin untuk mendaftarkan aplikasi dan memberikan izin API.

Pengaturan

Sebelum membuat pustaka klien Anda, instal dependensi yang diperlukan untuk membuat permintaan HTTP:

pip install requests

Implementasi pustaka klien

Buat kelas klien yang dapat digunakan kembali yang membungkus panggilan HTTP ke Microsoft Entra SDK untuk AgentID. Kelas ini menangani penerusan token, konfigurasi permintaan, dan penanganan kesalahan:

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

Integrasi Flask

Integrasikan perpustakaan klien ke dalam aplikasi Flask dengan mengekstrak token masuk dalam fungsi bantu dan menggunakannya di pengendali rute untuk memanggil API hilir.

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)

Integrasi FastAPI

Untuk aplikasi FastAPI, gunakan sistem injeksi dependensi dengan dependensi Header untuk mengekstrak dan memvalidasi token otorisasi sebelum meneruskannya ke pengelola rute.

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

Integrasi Django

Untuk aplikasi Django, buat kelas tampilan yang mengekstrak token otorisasi dari header permintaan dan gunakan untuk memanggil API hilir:

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

Tingkat lanjut: menggunakan permintaan. Sesi

Untuk meningkatkan performa dan ketahanan, gunakan requests.Session objek dengan logika coba lagi. Pendekatan ini memungkinkan percobaan ulang otomatis untuk kegagalan sementara dan pengumpulan koneksi untuk mengurangi overhead:

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

Praktik terbaik

Saat menggunakan Microsoft Entra SDK untuk AgentID dari Python, ikuti praktik ini untuk membangun aplikasi yang andal dan dapat dipertahankan:

Gunakan Kembali Instans Klien: Buat satu SidecarClient instans dan gunakan kembali di seluruh aplikasi Anda daripada membuat instans baru per permintaan. Ini meningkatkan performa dan penggunaan sumber daya.
Atur Batas Waktu yang Sesuai: Konfigurasikan batas waktu permintaan berdasarkan latensi API hilir Anda. Ini mencegah aplikasi Anda menggantung tanpa batas waktu jika SDK atau layanan hilir lambat.
Terapkan Penanganan Kesalahan: Tambahkan penanganan kesalahan yang tepat dan coba lagi logika, terutama untuk kegagalan sementara. Membedakan antara kesalahan klien (4xx) dan kesalahan server (5xx) untuk menentukan respons yang sesuai.
Gunakan Petunjuk Jenis: Tambahkan petunjuk jenis ke parameter fungsi dan kembalikan nilai untuk kejelasan kode yang lebih baik dan untuk menangkap kesalahan pada waktu pengembangan.
Aktifkan Pengumpulan Koneksi: Gunakan requests.Session objek untuk penggunaan kembali koneksi di seluruh permintaan, yang mengurangi overhead dan meningkatkan throughput untuk beberapa panggilan API.

Panduan bahasa lain

Langkah selanjutnya

Mulailah dengan skenario:

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2025-11-14