创建一个 Python 客户端库，该库与 Microsoft Entra SDK for AgentID 集成，以获取令牌并调用下游 API。 然后将此客户端集成到 Flask、FastAPI 或 Django 应用程序以处理经过身份验证的请求。
先决条件
- 拥有有效订阅的 Azure 帐户。 免费创建帐户。
- Python（版本 3.7 或更高版本）与 pip 已安装在您的开发计算机上。
- 在您的环境中部署和运行的 Microsoft Entra SDK for AgentID。 有关安装说明，请参阅 安装指南 。
- 在 SDK 中配置了基 URL 和所需范围的下游 API。
- 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 调用的吞吐量。
其他语言指南
后续步骤
从场景开始：