Implemente un agente MCP con OAuth de usuario final para que cada usuario se autentique con su propia identidad y el agente acceda solo a los recursos que el usuario ha autorizado. Las implementaciones de MCP predeterminadas utilizan una única cuenta de servicio, lo que significa que todos los usuarios comparten los mismos permisos y límites de velocidad. OAuth para el usuario final proporciona a cada usuario su propio token, lo que permite la limitación de tasas por usuario, seguimientos de auditoría y acceso a datos específicos. Esto es necesario para los agentes multiinquilino que acceden a datos específicos del usuario, como correo electrónico, calendario o repositorios privados.
Requisitos previos
- Python 3.8+ instalado
- Una implementación de servidor MCP (FastAPI, Express o similar)
- Un proveedor de OAuth 2.0 (Google, GitHub o personalizado)
- Una clave API de Scavio de scavio.dev para la herramienta de búsqueda
Guia paso a paso
Paso 1: Configurar ámbitos de OAuth
Defina los alcances de OAuth que su agente necesita y asígnelos a los permisos de la herramienta MCP.
Python
import os, json
# OAuth configuration
OAUTH_CONFIG = {
'provider': 'google', # or github, custom
'client_id': os.environ.get('OAUTH_CLIENT_ID', ''),
'client_secret': os.environ.get('OAUTH_CLIENT_SECRET', ''),
'redirect_uri': 'http://localhost:3000/callback',
'scopes': [
'openid',
'profile',
'email',
],
}
# Map scopes to tool permissions
TOOL_PERMISSIONS = {
'web_search': {'required_scopes': ['openid']},
'email_search': {'required_scopes': ['openid', 'email']},
'calendar_read': {'required_scopes': ['openid', 'https://www.googleapis.com/auth/calendar.readonly']},
}
print(f'Configured {len(TOOL_PERMISSIONS)} tools with scope requirements')Paso 2: Establecer el tipo de conexión como usuario final
Configure el servidor MCP para utilizar el tipo de conexión de usuario final en lugar de la cuenta de servicio.
Python
# MCP server configuration for end-user OAuth
mcp_config = {
'mcpServers': {
'search-agent': {
'url': 'http://localhost:3000/mcp',
'connection': {
'type': 'end_user',
'oauth': {
'provider': OAUTH_CONFIG['provider'],
'client_id': OAUTH_CONFIG['client_id'],
'scopes': OAUTH_CONFIG['scopes'],
},
},
'tools': list(TOOL_PERMISSIONS.keys()),
}
}
}
# Write config
with open('.mcp.json', 'w') as f:
json.dump(mcp_config, f, indent=2)
print('MCP config written with end-user OAuth')Paso 3: Implementar el agente con validación de token
Cree el punto final del agente que valide el token OAuth del usuario antes de ejecutar las herramientas.
Python
import requests
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
def validate_token(token: str) -> dict:
"""Validate OAuth token and extract user info."""
resp = requests.get('https://www.googleapis.com/oauth2/v3/userinfo',
headers={'Authorization': f'Bearer {token}'}, timeout=10)
if resp.status_code != 200:
return {'valid': False, 'error': 'invalid_token'}
user = resp.json()
return {'valid': True, 'email': user.get('email', ''), 'sub': user.get('sub', '')}
def execute_tool(tool_name: str, params: dict, user_token: str) -> dict:
user = validate_token(user_token)
if not user['valid']:
return {'error': 'unauthorized', 'message': 'Invalid or expired token'}
if tool_name == 'web_search':
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY},
json={'platform': 'google', 'query': params.get('query', '')}, timeout=10)
return {'results': resp.json().get('organic_results', [])[:5], 'user': user['email']}
return {'error': f'Unknown tool: {tool_name}'}
print('Agent endpoint ready with token validation')Paso 4: Prueba con identidad de usuario final
Simule una solicitud de usuario final y verifique que el agente utilice correctamente la identidad del usuario.
Python
def test_end_user_flow():
# Simulate with a mock token (in production, this comes from OAuth flow)
mock_token = 'mock_user_token_for_testing'
# Test the search tool
result = execute_tool('web_search', {'query': 'python best practices 2026'}, mock_token)
if 'error' in result and result['error'] == 'unauthorized':
print('Token validation working (mock token rejected as expected)')
else:
print(f'Tool executed for user: {result.get("user", "unknown")}')
# Test with Scavio directly (no OAuth needed)
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY},
json={'platform': 'google', 'query': 'oauth 2.0 best practices'}, timeout=10)
results = resp.json().get('organic_results', [])
print(f'Direct API test: {len(results)} results')
test_end_user_flow()Paso 5: Verificar el aislamiento de identidad por usuario
Confirme que las solicitudes de cada usuario estén aisladas y rastreadas por separado.
Python
from collections import defaultdict
user_audit = defaultdict(list)
def audited_tool(tool_name: str, params: dict, user_email: str) -> dict:
"""Execute tool with per-user audit logging."""
user_audit[user_email].append({
'tool': tool_name,
'params': params,
'timestamp': __import__('time').time(),
})
if tool_name == 'web_search':
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY},
json={'platform': 'google', 'query': params.get('query', '')}, timeout=10)
return {'results': resp.json().get('organic_results', [])[:3]}
return {}
# Simulate multiple users
audited_tool('web_search', {'query': 'react tutorial'}, '[email protected]')
audited_tool('web_search', {'query': 'vue tutorial'}, '[email protected]')
audited_tool('web_search', {'query': 'angular tutorial'}, '[email protected]')
for user, actions in user_audit.items():
print(f'{user}: {len(actions)} tool calls')
print('Per-user isolation verified')Ejemplo en Python
Python
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def user_search(query, user_email):
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
results = data.get('organic_results', [])[:3]
print(f'[{user_email}] {query}: {len(results)} results')
return results
user_search('python oauth tutorial', '[email protected]')Ejemplo en JavaScript
JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function userSearch(query, userEmail) {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
const results = (await r.json()).organic_results || [];
console.log(`[${userEmail}] ${query}: ${results.length} results`);
return results.slice(0, 3);
}
userSearch('oauth best practices', '[email protected]');Salida esperada
JSON
An MCP agent deployed with end-user OAuth where each user authenticates independently, tool calls are scoped to user permissions, and all actions are audited per user.