Microsoft Entra SDK for AgentID と統合してトークンを取得し、ダウンストリーム API を呼び出す Python クライアント ライブラリを作成します。 次に、このクライアントを Flask、FastAPI、または Django アプリケーションに統合して、認証された要求を処理します。
[前提条件]
- アクティブなサブスクリプションを持つ Azure アカウント。 無料でアカウントを作成できます。
- 開発用コンピューターに pip がインストールされている Python (バージョン 3.7 以降)。
- お使いの環境でデプロイおよび実行されている Microsoft Entra SDK for AgentID。 セットアップ手順については 、インストール ガイド を参照してください。
- ベース URL と必要なスコープを使用して SDK で構成されたダウンストリーム 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 の統合
ヘルパー関数で受信トークンを抽出し、それをルート ハンドラーで使用してダウンストリーム API を呼び出すことで、クライアント ライブラリを Flask アプリケーションに統合します。
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 から Microsoft Entra SDK for AgentID を使用する場合は、次のプラクティスに従って、信頼性と保守性に優れたアプリケーションを構築します。
-
クライアント インスタンスの再利用: 1 つの
SidecarClientインスタンスを作成し、要求ごとに新しいインスタンスを作成するのではなく、アプリケーション全体で再利用します。 これにより、パフォーマンスとリソースの使用量が向上します。 - 適切なタイムアウトの設定: ダウンストリーム API の待機時間に基づいて要求タイムアウトを構成します。 これにより、SDK またはダウンストリーム サービスの速度が低下した場合に、アプリケーションが無期限にハングすることを防ぐことができます。
- エラー処理を実装する: 特に一時的なエラーの場合は、適切なエラー処理と再試行ロジックを追加します。 クライアント エラー (4xx) とサーバー エラー (5xx) を区別して、適切な応答を判断します。
- 型ヒントを使用する: コードをわかりやすくし、開発時にエラーをキャッチするために、関数パラメーターと戻り値に型ヒントを追加します。
-
接続プールを有効にする: 要求間で接続を再利用するために
requests.Sessionオブジェクトを使用します。これにより、オーバーヘッドが軽減され、複数の API 呼び出しのスループットが向上します。
その他の言語ガイド
次のステップ
次のシナリオから始めます。