Sistema Operativo de Economía de Agentes Universal (UAE OS)

El Sistema Operativo de Economía de Agentes Universal es una plataforma central fundamental, nativa de MCP/A2A, diseñada para impulsar la explosiva subeconomía de agentes. Comienza como un proxy seguro de inyección de credenciales y micropagos x402, que se convierte diariamente en un imperio de monopolio múltiple (motor de identidad, pagos, liquidación, paquetes de cumplimiento, mercados verticales).

Este esqueleto de proxy v0 es la base inquebrantable que cada módulo futuro ampliará. Es totalmente compatible con el Protocolo de Contexto de Modelo (MCP) y Agente a Agente (A2A), lo que permite a los agentes autónomos descubrir, autenticar y pagarse entre sí de forma segura sin intervención humana.

Inicio rápido con el SDK de Python

La forma más fácil de interactuar con el UAE OS es a través del SDK oficial de Python. Proporciona agrupación de conexiones robusta, reintentos automáticos con retroceso exponencial para errores transitorios (429s, 5xxs) y manejo estructurado de excepciones que se asigna directamente al sistema UAEError.

1. Instalación

Instala el SDK directamente desde la raíz del repositorio:

pip install -e .

2. Uso básico y manejo de errores

import asyncio
from sdk.uaeos import UAEOSClient
from sdk.uaeos.client import RateLimitError, AuthError, InsufficientScopesError, APIError

async def main():
    # Initialize the client with your API key (uses async context manager for connection pooling)
    # The client automatically retries transient errors (max_retries=3 by default)
    async with UAEOSClient(api_key="sk_test_1234567890abcdef", base_url="http://127.0.0.1:8000") as client:
        try:
            # 1. Register a new agent in the Identity Engine
            agent = await client.register_agent(
                agent_id="agent_sdk_1", 
                name="SDK Test Agent", 
                metadata={"version": "1.0"}
            )
            print("Registered:", agent)
            
            # 2. Rotate/Issue a credential with cryptographic scopes
            cred = await client.rotate_credential(
                agent_id="agent_sdk_1",
                credential_type="stripe_live",
                new_secret_data={"api_key": "sk_live_new123"},
                expires_in_days=30
            )
            print("Credential Rotated:", cred)
            
            # 3. Execute an MCP/A2A tool call with x402 micropayment
            result = await client.execute(
                agent_id="agent_sdk_1",
                tool_call={
                    "target_agent_id": "agent_target_2",
                    "action": "process_data",
                    "payload": {"hello": "world"},
                    "required_scopes": ["read"]
                },
                credential_type="stripe_live",
                payment_amount=1.50
            )
            print("Execution Result:", result)
            
            # 4. Execute a direct A2A payment (no downstream HTTP call)
            payment_result = await client.execute_payment(
                agent_id="agent_sdk_1",
                payment_amount=5.00,
                target_agent_id="agent_target_3",
                action="data_purchase"
            )
            print("Payment Result:", payment_result)
            
            # 5. Get Usage Stats from the Analytics Engine
            stats = await client.get_stats()
            print("Global Stats:", stats)
            
            # 6. Generate a usage-based invoice for the agent
            invoice = await client.get_invoice(agent_id="agent_sdk_1")
            print("Generated Invoice:", invoice)
            
        except RateLimitError as e:
            # Raised if the agent exceeds the rate limit and max_retries are exhausted
            print(f"Rate limited! Retry after {e.retry_after} seconds. Request ID: {e.request_id}")
        except InsufficientScopesError as e:
            # Raised if the agent lacks the required scopes for the credential
            print(f"Permission denied: {e.message}")
        except AuthError:
            # Raised for 401 Unauthorized
            print("Invalid API Key!")
        except APIError as e:
            # Catch-all for other 4xx/5xx errors or network issues
            print(f"API Error ({e.status_code}): {e.message}")
        except Exception as e:
            print(f"An unexpected error occurred: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Notas de la versión v0.1

¡Bienvenido al lanzamiento fundamental del Sistema Operativo de Economía de Agentes Universal! Durante los últimos 30 días, hemos construido un esqueleto de proxy altamente modular y listo para producción que actúa como el enrutador central para la red MCP/A2A.

Características en v0.1.0:

• FastAPI + Pydantic v2 Core: Puerta de enlace API de alto rendimiento y tipado estricto.

• Motor de Identidad: Integración con Supabase para búsqueda, inyección, rotación y validación de alcance criptográfico de credenciales de forma segura.

• Motor de Liquidación: Manejo de micropagos x402, verificación de webhooks de Stripe/Lightning y generación de facturas de facturación basadas en el uso.

• Enrutamiento A2A: Stub de enrutamiento inteligente de Agente a Agente junto con ejecución descendente (httpx).

• Paquetes de Cumplimiento: Registro de auditoría y generación de ID adt_ únicos.

• Control de Tráfico: Limitación de tasa lista para Redis (10 req/min) con respuestas 429 adecuadas.

• Caché: Capa de caché de identidad lista para Redis para credenciales y alcances.

• Seguridad: Autenticación de clave API (Authorization: Bearer y X-API-Key), middleware CORS, encabezados de seguridad personalizados e informes de errores estructurados (UAEError).

• Panel de Análisis de Uso: Seguimiento de uso en memoria seguro para subprocesos, registro de actividad reciente y un panel global /stats.

• SDK de Python: Un cliente oficial de Python centrado en async (UAEOSClient) con agrupación de conexiones, reintentos con retroceso exponencial y manejo estructurado de errores.

• Configuración: Configuración centralizada de Pydantic (app/config.py) como única fuente de verdad.

• Listo para Despliegue: Dockerizado y optimizado para despliegue en Railway con un solo clic.

• 100% de Cobertura de Pruebas: Suite de pruebas integral y totalmente simulada con 82 pruebas de integración.

Inicio rápido (Servidor local)

1. Instalar dependencias

pip install -r requirements.txt

2. Configurar entorno Copia .env.example a .env y completa tus detalles de Supabase (opcional para simulación local).

cp .env.example .env

3. Ejecutar el servidor

uvicorn app.main:app --reload

La API estará disponible en http://127.0.0.1:8000. Consulta http://127.0.0.1:8000/docs para ver la interfaz interactiva de Swagger.

4. Ejecutar pruebas

pytest -v

Inicio rápido (Docker)

Para ejecutar el proxy en un contenedor aislado:

1. Construir la imagen

docker build -t uae-os-proxy .

2. Ejecutar el contenedor

docker run -p 8000:8000 --env-file .env uae-os-proxy

Variables de entorno

La aplicación se configura completamente mediante variables de entorno (gestionadas por app/config.py). Consulta .env.example para obtener más detalles.

Nota: Si faltan las variables de Supabase, la aplicación vuelve a un modo de simulación para pruebas.

Ejemplos de uso de API (cURL)

El endpoint del proxy (/proxy/execute) está protegido y requiere tu API_KEY. Puedes autenticarte usando el encabezado estándar Authorization: Bearer <key> o el encabezado X-API-Key: <key>.

1. Llamada de herramienta básica (sin pago)

Usando cURL (Bearer Token):

curl -X POST http://127.0.0.1:8000/proxy/execute \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer sk_test_1234567890abcdef" \
  -d '{
    "agent_id": "agent_alpha",
    "tool_call": {
      "url": "https://api.example.com/v1/data",
      "method": "POST",
      "payload": {"query": "test"}
    },
    "credential_type": "stripe_live"
  }'

2. Llamada de herramienta MCP/A2A con micropago x402

Para simular una liquidación y enrutar a otro agente, incluye los campos payment_amount y target_agent_id.

curl -X POST http://127.0.0.1:8000/proxy/execute \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer sk_test_1234567890abcdef" \
  -d '{
    "agent_id": "agent_beta",
    "tool_call": {
      "target_agent_id": "agent_gamma",
      "action": "premium_data_fetch"
    },
    "credential_type": "custom_oauth",
    "payment_amount": 0.50
  }'

Respuesta esperada:

{
  "success": true,
  "injected_credential": true,
  "x402_settled": true,
  "transaction_id": "tx_a1b2c3d4e5f6",
  "audit_id": "adt_9876543210abcdef"
}

Despliegue (Costo cero / Un clic)

El Dockerfile está optimizado para proveedores PaaS modernos. Utiliza una imagen ligera de Python 3.11 slim, expone el puerto 8000 y se vincula a 0.0.0.0.

Despliegue en Railway (Recomendado)

Railway ofrece una experiencia de despliegue fluida con un solo clic que lee automáticamente el railway.toml y el Dockerfile incluidos.

1. Push a GitHub: Haz commit de este repositorio a un repositorio de GitHub.

2. Conectar Railway: Inicia sesión en Railway y haz clic en New Project -> Deploy from GitHub repo.

3. Seleccionar repositorio: Elige tu repositorio recién subido.

4. Establecer variables de entorno: En el panel de tu proyecto de Railway, ve a la pestaña Variables y agrega tu API_KEY, WEBHOOK_SECRET y cualquier clave opcional de Supabase/Stripe. Asegúrate de generar cadenas aleatorias fuertes para los secretos en producción.

5. Desplegar: Railway detectará automáticamente el archivo railway.toml, construirá la imagen de Docker y la desplegará.

Patrón de URL en vivo esperado: https://agent-economy-os-production.up.railway.app

Monitoreo y verificación de producción

Una vez desplegado, puedes verificar que el servicio se está ejecutando y monitorear su estado usando los endpoints públicos. Estos son los endpoints recomendados para usar en las comprobaciones de salud de PaaS:

• Comprobación de salud: https://agent-economy-os-production.up.railway.app/health

  • Devuelve el estado básico, la versión y la marca de tiempo.

• Métricas: https://agent-economy-os-production.up.railway.app/metrics

  • Devuelve el tiempo de actividad y el total de agentes medidos.

Nota: Dado que la limitación de tasa y el almacenamiento en caché actuales están en memoria, asegúrate de que tu PaaS esté configurado para ejecutar una sola instancia/réplica (que es el valor predeterminado en los niveles gratuitos) hasta que Redis esté completamente integrado.

Configuración de Webhook de Stripe

Para habilitar el seguimiento de liquidación en tiempo real para pagos de Stripe en vivo, configura un webhook en tu Panel de Stripe:

1. Ve a Developers > Webhooks en el Panel de Stripe.

2. Haz clic en Add endpoint.

3. Establece la Endpoint URL en https://agent-economy-os-production.up.railway.app/webhooks/stripe.

4. Selecciona los Events to send: payment_intent.succeeded y payment_intent.payment_failed.

5. Haz clic en Add endpoint.

6. Revela el Signing secret (comienza con whsec_) y agrégalo a tus variables de entorno de Railway como STRIPE_WEBHOOK_SECRET.

Contribución

¡Damos la bienvenida a las contribuciones! El Sistema Operativo de Economía de Agentes Universal está diseñado para ser el estándar definitivo de código abierto para la economía de agentes.

Asegúrate de que todas las pruebas pasen (pytest -v) y que la cobertura se mantenga al 100% antes de enviar una Solicitud de Extracción (Pull Request). Si estás agregando un nuevo módulo, asegúrate de que se base en la arquitectura existente sin romper el flujo de ejecución del proxy central.