Universal Agent Economy OS (UAE OS)

Universal Agent Economy OS — это фундаментальная, MCP/A2A-ориентированная базовая платформа, разработанная для поддержки стремительно развивающейся агентной субакономики. Она начинает свою работу как прокси для безопасного внедрения учетных данных и микроплатежей x402, ежедневно превращаясь в полноценную мультимонопольную империю (движок идентификации, платежи, расчеты, пакеты соответствия требованиям, вертикальные маркетплейсы).

Этот прокси-скелет версии v0 является нерушимым фундаментом, который будет расширяться каждым будущим модулем. Он полностью совместим с Model Context Protocol (MCP) и Agent-to-Agent (A2A), позволяя автономным агентам безопасно находить, аутентифицировать и оплачивать услуги друг друга без вмешательства человека.

Быстрый старт с Python SDK

Самый простой способ взаимодействия с UAE OS — через официальный Python SDK. Он обеспечивает надежное пулирование соединений, автоматические повторные попытки с экспоненциальной задержкой при временных ошибках (429, 5xx) и структурированную обработку исключений, которая напрямую сопоставляется с системой UAEError.

1. Установка

Установите SDK прямо из корня репозитория:

pip install -e .

2. Базовое использование и обработка ошибок

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())

Примечания к релизу v0.1

Добро пожаловать в фундаментальный релиз Universal Agent Economy OS! За последние 30 дней мы создали высокомодульный, готовый к продакшену прокси-скелет, который выступает в качестве основного маршрутизатора для сети MCP/A2A.

Функции в v0.1.0:

• FastAPI + Pydantic v2 Core: Строго типизированный высокопроизводительный API-шлюз.

• Движок идентификации: Интеграция с Supabase для безопасного поиска, внедрения, ротации учетных данных и криптографической проверки областей действия (scopes).

• Движок расчетов: Обработка микроплатежей x402, проверка вебхуков Stripe/Lightning и генерация счетов на основе использования.

• Маршрутизация A2A: Заглушка для интеллектуальной маршрутизации «агент-агент» вместе с последующим выполнением (httpx).

• Пакеты соответствия: Журналирование аудита и генерация уникальных ID adt_.

• Контроль трафика: Ограничение частоты запросов (10 запросов/мин) с поддержкой Redis и корректными ответами 429.

• Кэширование: Уровень кэширования идентификационных данных с поддержкой Redis для учетных данных и областей действия.

• Безопасность: Аутентификация по API-ключу (Authorization: Bearer и X-API-Key), CORS-middleware, пользовательские заголовки безопасности и структурированные отчеты об ошибках (UAEError).

• Панель аналитики использования: Потокобезопасное отслеживание использования в памяти, журнал недавней активности и глобальная панель /stats.

• Python SDK: Официальный Python-клиент с поддержкой асинхронности (UAEOSClient), пулированием соединений, повторными попытками с экспоненциальной задержкой и структурированной обработкой ошибок.

• Конфигурация: Централизованные настройки Pydantic (app/config.py) как единственный источник истины.

• Готовность к развертыванию: Контейнеризация Docker и оптимизация для развертывания в Railway в один клик.

• 100% покрытие тестами: Полностью мокированный, комплексный набор тестов из 82 интеграционных тестов.

Быстрый старт (Локальный сервер)

1. Установка зависимостей

pip install -r requirements.txt

2. Настройка окружения Скопируйте .env.example в .env и заполните данные Supabase (опционально для локальной симуляции).

cp .env.example .env

3. Запуск сервера

uvicorn app.main:app --reload

API будет доступен по адресу http://127.0.0.1:8000. Проверьте http://127.0.0.1:8000/docs для интерактивного Swagger UI.

4. Запуск тестов

pytest -v

Быстрый старт (Docker)

Чтобы запустить прокси в изолированном контейнере:

1. Сборка образа

docker build -t uae-os-proxy .

2. Запуск контейнера

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

Переменные окружения

Приложение полностью настраивается через переменные окружения (управляются через app/config.py). Подробности см. в .env.example.

Примечание: Если переменные Supabase отсутствуют, приложение корректно переключается в режим симуляции для тестирования.

Примеры использования API (cURL)

Эндпоинт прокси (/proxy/execute) защищен и требует ваш API_KEY. Вы можете пройти аутентификацию, используя заголовок Authorization: Bearer <key> или X-API-Key: <key>.

1. Базовый вызов инструмента (без оплаты)

Использование 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. Вызов инструмента MCP/A2A с микроплатежом x402

Чтобы симулировать расчет и маршрутизацию к другому агенту, включите поля payment_amount и 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
  }'

Ожидаемый ответ:

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

Развертывание (Бесплатно / В один клик)

Dockerfile оптимизирован для современных PaaS-провайдеров. Он использует легкий образ Python 3.11 slim, открывает порт 8000 и привязывается к 0.0.0.0.

Развертывание в Railway (Рекомендуется)

Railway предлагает бесшовное развертывание в один клик, которое автоматически считывает включенные railway.toml и Dockerfile.

1. Push в GitHub: Закоммитьте этот репозиторий в репозиторий GitHub.

2. Подключение Railway: Войдите в Railway и нажмите New Project -> Deploy from GitHub repo.

3. Выбор репозитория: Выберите ваш недавно запушенный репозиторий.

4. Установка переменных окружения: В панели управления проектом Railway перейдите на вкладку Variables и добавьте API_KEY, WEBHOOK_SECRET и любые опциональные ключи Supabase/Stripe. Убедитесь, что вы генерируете надежные случайные строки для секретов в продакшене.

5. Развертывание: Railway автоматически обнаружит файл railway.toml, соберет Docker-образ и развернет его.

Ожидаемый шаблон URL: https://agent-economy-os-production.up.railway.app

Мониторинг и проверка продакшена

После развертывания вы можете проверить работу сервиса и отслеживать его состояние с помощью публичных эндпоинтов. Это рекомендуемые эндпоинты для проверок работоспособности PaaS:

• Проверка работоспособности: https://agent-economy-os-production.up.railway.app/health

  • Возвращает базовый статус, версию и временную метку.

• Метрики: https://agent-economy-os-production.up.railway.app/metrics

  • Возвращает время работы и общее количество измеренных агентов.

Примечание: Поскольку текущее ограничение частоты запросов и кэширование работают в памяти, убедитесь, что ваш PaaS настроен на запуск одного экземпляра/реплики (что является стандартом для бесплатных тарифов), пока Redis не будет полностью интегрирован.

Настройка вебхуков Stripe

Чтобы включить отслеживание расчетов в реальном времени для платежей Stripe, настройте вебхук в панели управления Stripe:

1. Перейдите в Developers > Webhooks в панели управления Stripe.

2. Нажмите Add endpoint.

3. Установите Endpoint URL на https://agent-economy-os-production.up.railway.app/webhooks/stripe.

4. Выберите Events to send: payment_intent.succeeded и payment_intent.payment_failed.

5. Нажмите Add endpoint.

6. Скопируйте Signing secret (начинается с whsec_) и добавьте его в переменные окружения Railway как STRIPE_WEBHOOK_SECRET.

Вклад в проект

Мы приветствуем любой вклад! Universal Agent Economy OS создана как окончательный стандарт с открытым исходным кодом для агентной экономики.

Пожалуйста, убедитесь, что все тесты проходят (pytest -v) и покрытие остается на уровне 100% перед отправкой Pull Request. Если вы добавляете новый модуль, убедитесь, что он дополняет существующую архитектуру, не нарушая основной поток выполнения прокси.