Сценарий. Использование пакета SDK Microsoft Entra для идентификатора агента из Python

Создайте клиентскую библиотеку на языке Python, которая интегрируется с Microsoft Entra SDK для идентификатора агента, чтобы получать токены и вызывать API нижнего уровня. Затем интегрируйте этот клиент в приложения Flask, FastAPI или Django для обработки прошедших проверку подлинности запросов.

Предпосылки

Учетная запись Azure с активной подпиской. Создайте учетную запись бесплатно .
Python (версия 3.7 или более поздняя) с pip, установленным на вашей машине разработки.
пакет SDK Microsoft Entra для AgentID развернут и запущен в вашей среде. Инструкции по установке см. в руководстве по установке .
Подчиненные API, настроенные в пакете SDK с базовыми URL-адресами и необходимыми областями.
Соответствующие разрешения в Microsoft Entra ID — у вашей учетной записи должны быть разрешения на регистрацию приложений и предоставление разрешений API.

Настройка

Перед созданием клиентской библиотеки установите необходимую зависимость для выполнения HTTP-запросов:

pip install requests

Реализация клиентской библиотеки

Создайте повторно используемый клиентский класс, который упаковывает HTTP-вызовы в пакет SDK Microsoft Entra для AgentID. Этот класс обрабатывает пересылку токенов, конфигурацию запроса и обработку ошибок.

# 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 повторных попыток. Этот подход позволяет автоматически повторять кратковременные сбои и организовывать пул подключений, чтобы снизить накладные расходы:

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

Лучшие практики

При использовании пакета SDK Microsoft Entra для идентификатора агента из Python выполните следующие действия для создания надежных и обслуживаемых приложений:

Повторное использование экземпляра клиента: создайте один SidecarClient экземпляр и повторно используйте его во всем приложении, а не создавайте новые экземпляры на запрос. Это повышает производительность и использование ресурсов.
Задайте соответствующие значения времени ожидания: настройте время ожидания запроса на основе задержки нижестоящего API. Это предотвращает зависание вашего приложения на неопределенный срок, если SDK или зависимая служба работают медленно.
Реализуйте обработку ошибок: добавьте правильную обработку ошибок и логику повторных попыток, особенно для временных сбоев. Различаете ошибки клиента (4xx) и ошибки сервера (5xx) для определения соответствующих ответов.
Используйте подсказки типов: добавьте подсказки типов в параметры функции и возвращаемые значения для повышения четкости кода и перехвата ошибок во время разработки.
Включение пула подключений. Используйте requests.Session объект для повторного использования подключения между запросами, что снижает затраты и повышает пропускную способность для нескольких вызовов API.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-04-19