Observação
O acesso a essa página exige autorização. Você pode tentar entrar ou alterar diretórios.
O acesso a essa página exige autorização. Você pode tentar alterar os diretórios.
Crie uma biblioteca de clientes python que se integre ao SDK do Microsoft Entra para AgentID para obter tokens e chamar APIs downstream. Em seguida, integre esse cliente aos aplicativos Flask, FastAPI ou Django para lidar com solicitações autenticadas.
Pré-requisitos
- Uma conta do Azure com uma assinatura ativa. Crie uma conta gratuitamente.
- Python (versão 3.7 ou posterior) com pip instalado em seu computador de desenvolvimento.
- SDK do Microsoft Entra para AgentID implantado e em execução em seu ambiente. Consulte o Guia de Instalação para obter instruções de instalação.
- APIs downstream configuradas no SDK com URLs base e escopos necessários.
- Permissões apropriadas no Microsoft Entra ID – Sua conta deve ter permissões para registrar aplicativos e conceder permissões de API.
Configuração
Antes de criar sua biblioteca de clientes, instale a dependência necessária para fazer solicitações HTTP:
pip install requests
Implementação da biblioteca de clientes
Crie uma classe de cliente reutilizável que encapsula chamadas HTTP para o SDK do Microsoft Entra para AgentID. Essa classe manipula o encaminhamento de token, a configuração de solicitação e o tratamento de erros:
# 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')
Integração do Flask
Integre a biblioteca de clientes a um aplicativo Flask extraindo o token de entrada em uma função auxiliar e usando-o em manipuladores de rota para chamar APIs downstream:
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)
Integração do FastAPI
Para aplicativos FastAPI, use o sistema de injeção de dependência com a Header dependência para extrair e validar o token de autorização antes de passá-lo para manipuladores de rota:
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))
Integração do Django
Para aplicativos Django, crie classes de exibição que extraam o token de autorização de cabeçalhos de solicitação e use-o para chamar APIs downstream:
# 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)
Avançado: usando solicitações. Sessão
Para melhorar o desempenho e a resiliência, use um requests.Session objeto com lógica de repetição. Essa abordagem permite novas tentativas automáticas de falhas transitórias e pool de conexões para reduzir a sobrecarga:
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
Práticas recomendadas
Ao usar o SDK do Microsoft Entra para AgentID do Python, siga estas práticas para criar aplicativos confiáveis e mantêveis:
-
Reutilizar a Instância do Cliente: crie uma única
SidecarClientinstância e reutilize-a em todo o aplicativo, em vez de criar novas instâncias por solicitação. Isso melhora o desempenho e o uso de recursos. - Definir tempos limite apropriados: configure tempos limite de solicitação com base na latência da API downstream. Isso impede que seu aplicativo fique suspenso indefinidamente se o SDK ou o serviço downstream estiver lento.
- Implemente o Tratamento de Erros: adicione a lógica de repetição e tratamento de erros adequada, especialmente para falhas transitórias. Distingue entre erros de cliente (4xx) e erros de servidor (5xx) para determinar as respostas apropriadas.
- Use anotações de tipo: adicione anotações de tipo aos parâmetros das funções e valores de retorno para uma maior clareza de código e para detectar erros durante o desenvolvimento.
-
Habilitar o Pool de Conexões: use um
requests.Sessionobjeto para reutilização de conexão entre solicitações, o que reduz a sobrecarga e melhora a taxa de transferência para várias chamadas à API.
Outros guias de idioma
Próximas etapas
Comece com um cenário: