案例：從 Python 使用適用於 AgentID 的 Microsoft Entra SDK

建立與適用於 AgentID 的 Microsoft Entra SDK 整合的 Python 用戶端程式庫，以取得權杖並呼叫下游 API。然後將此客戶端整合到 Flask、FastAPI 或 Django 應用程式中，以處理經過身份驗證的請求。

先決條件

具有有效訂閱的 Azure 帳戶。免費建立帳戶。
Python （3.7 版或更新版本），並在您的開發機器上安裝了 pip。
適用於 AgentID 的 Microsoft Entra SDK 已在您的環境中部署並執行。如需設定指示，請參閱安裝指南。
在 SDK 中設定的下游 API，具有基底 URL 和必要範圍。
Microsoft Entra ID 中的適當許可權 - 您的帳戶必須具有註冊應用程式和授與 API 許可權的許可權。

設定

在建立用戶端程式庫之前，請安裝發出 HTTP 要求所需的相依性：

pip install requests

用戶端程式庫實作

建立可重複使用的用戶端類別，以包裝對 Microsoft Entra SDK for AgentID 的 HTTP 呼叫。這個類別處理權杖轉送、請求配置和錯誤處理：

# 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 整合

將客戶端程式庫整合到 Flask 應用程式中，可以透過在輔助函式中擷取進入的權杖，並在路由處理程序中使用此權杖來呼叫下游 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)

FastAPI 整合

對於 FastAPI 應用程式，請使用相依性插入系統 Header 來擷取和驗證授權令牌，再將其傳遞至路由處理器之前。

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 整合

對於 Django 應用程序，建立視圖類別，從請求標頭中提取授權令牌，並使用它來呼叫下游 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)

進階：使用 requests.Session

若要改善效能和復原能力，請使用 requests.Session 具有重試邏輯的物件。此方法支援針對暫時性失敗自動重試，並透過連線資源池來降低額外負荷。

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

最佳做法

從 Python 使用適用於 AgentID 的 Microsoft Entra SDK 時，請遵循下列做法來建置可靠且可維護的應用程式：

重複使用用戶端執行個體：建立單一 SidecarClient 執行個體並在整個應用程式中重複使用它，而不是為每個請求建立新執行個體。這可改善效能和資源使用量。
設置適當的超時： 根據您的下游 API 延遲配置請求超時。這可防止您的應用程式在 SDK 或下游服務速度緩慢時無限期停止回應。
實作錯誤處理：新增適當的錯誤處理和重試邏輯，特別是對於暫時性失敗。區分用戶端錯誤（4xx）和伺服器錯誤（5xx）以判斷適當的回應。
使用類型提示： 將類型提示添加到函數參數和返回值中，以提高代碼清晰度並在開發時捕獲錯誤。
啟用連線池：使用物件 requests.Session 跨請求重複使用連線，從而減少開銷並提高多個 API 呼叫的吞吐量。

其他語言指南

後續步驟

從一個場景開始：

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-11-14