토큰을 가져오고 다운스트림 API를 호출하기 위해 AgentID용 Microsoft Entra SDK와 통합되는 Python 클라이언트 라이브러리를 만듭니다. 그런 다음 이 클라이언트를 Flask, FastAPI 또는 Django 애플리케이션에 통합하여 인증된 요청을 처리합니다.
필수 조건
- 활성 구독이 있는 Azure 계정. 무료로 계정을 만듭니다.
- 개발 머신에 pip가 설치된 Python(버전 3.7 이상)
- 사용자 환경에서 배포되고 실행되는 AgentID용 Microsoft Entra SDK입니다. 설치 지침은 설치 가이드 를 참조하세요.
- 기본 URL 및 필수 범위를 사용하여 SDK에 구성된 다운스트림 API입니다.
- Microsoft Entra ID의 적절한 권한 - 계정에 애플리케이션을 등록하고 API 권한을 부여할 수 있는 권한이 있어야 합니다.
설치 프로그램
클라이언트 라이브러리를 만들기 전에 HTTP 요청을 만드는 데 필요한 종속성을 설치합니다.
pip install requests
클라이언트 라이브러리 구현
AgentID용 Microsoft Entra SDK에 대한 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 사용합니다. 이 방법을 사용하면 일시적인 오류 및 연결 풀링에 대한 자동 재시도를 통해 오버헤드를 줄일 수 있습니다.
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)를 구분하여 적절한 응답을 확인합니다.
- 형식 힌트 사용: 함수 매개 변수에 형식 힌트를 추가하고 코드 명확성을 높이고 개발 시 오류를 catch하기 위해 값을 반환합니다.
-
연결 풀링 사용: 요청 간에 연결을 다시 사용하기 위해 개체를 사용하여
requests.Session오버헤드를 줄이고 여러 API 호출에 대한 처리량을 향상시킵니다.
기타 언어 가이드
다음 단계
시나리오 시작: