Universal Agent Economy OS (UAE OS)

Das Universal Agent Economy OS ist eine grundlegende, MCP/A2A-native Kernplattform, die entwickelt wurde, um die explodierende agentische Subökonomie anzutreiben. Es beginnt als sicherer Proxy für Anmeldedaten-Injektion und x402-Mikrozahlungen und entwickelt sich täglich zu einem vollständigen Multi-Monopol-Imperium (Identitäts-Engine, Zahlungen, Abrechnung, Compliance-Pakete, vertikale Marktplätze) weiter.

Dieses v0-Proxy-Skelett ist das unzerbrechliche Fundament, das jedes zukünftige Modul erweitern wird. Es ist vollständig Model Context Protocol (MCP)- und Agent-to-Agent (A2A)-kompatibel, was es autonomen Agenten ermöglicht, sich gegenseitig sicher zu entdecken, zu authentifizieren und zu bezahlen, ohne menschliches Eingreifen.

Schnellstart mit dem Python SDK

Der einfachste Weg, mit dem UAE OS zu interagieren, ist über das offizielle Python SDK. Es bietet robustes Connection-Pooling, automatische exponentielle Backoff-Wiederholungen bei transienten Fehlern (429er, 5xxer) und eine strukturierte Fehlerbehandlung, die direkt auf das UAEError-System abgebildet wird.

1. Installation

Installieren Sie das SDK direkt aus dem Repository-Root:

pip install -e .

2. Grundlegende Verwendung & Fehlerbehandlung

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 Release Notes

Willkommen beim grundlegenden Release des Universal Agent Economy OS! In den letzten 30 Tagen haben wir ein hochgradig modulares, produktionsreifes Proxy-Skelett gebaut, das als Kern-Router für das MCP/A2A-Netzwerk fungiert.

Funktionen in v0.1.0:

• FastAPI + Pydantic v2 Core: Streng typisiertes, leistungsstarkes API-Gateway.

• Identitäts-Engine: Supabase-Integration für sichere Suche, Injektion, Rotation und kryptografische Gültigkeitsprüfung von Anmeldedaten.

• Abrechnungs-Engine: x402-Mikrozahlungsabwicklung, Stripe/Lightning-Webhook-Verifizierung und nutzungsbasierte Rechnungserstellung.

• A2A-Routing: Intelligenter Agent-zu-Agent-Routing-Stub neben der nachgelagerten Ausführung (httpx).

• Compliance-Pakete: Audit-Logging und Generierung eindeutiger adt_-IDs.

• Traffic-Kontrolle: Redis-fähiges Rate Limiting (10 Anfragen/Min) mit korrekten 429-Antworten.

• Caching: Redis-fähige Identitäts-Caching-Schicht für Anmeldedaten und Scopes.

• Sicherheit: API-Key-Authentifizierung (Authorization: Bearer und X-API-Key), CORS-Middleware, benutzerdefinierte Sicherheits-Header und strukturierte Fehlerberichterstattung (UAEError).

• Nutzungsanalyse-Dashboard: Thread-sichere In-Memory-Nutzungsverfolgung, Protokollierung der letzten Aktivitäten und ein globales /stats-Dashboard.

• Python SDK: Ein offizieller, Async-First Python-Client (UAEOSClient) mit Connection-Pooling, exponentiellen Backoff-Wiederholungen und strukturierter Fehlerbehandlung.

• Konfiguration: Zentralisierte Pydantic-Einstellungen (app/config.py) als einzige Quelle der Wahrheit.

• Bereit für den Einsatz: Dockerisiert und optimiert für Railway-Deployment mit einem Klick.

• 100% Testabdeckung: Vollständig gemockte, umfassende Testsuite mit 82 Integrationstests.

Schnellstart (Lokaler Server)

1. Abhängigkeiten installieren

   pip install -r requirements.txt

2. Umgebung konfigurieren Kopieren Sie .env.example nach .env und geben Sie Ihre Supabase-Details ein (optional für lokale Simulation).

   cp .env.example .env

3. Server starten

   uvicorn app.main:app --reload

   Die API ist unter http://127.0.0.1:8000 verfügbar. Überprüfen Sie http://127.0.0.1:8000/docs für die interaktive Swagger-UI.

4. Tests ausführen

   pytest -v

Schnellstart (Docker)

Um den Proxy in einem isolierten Container auszuführen:

1. Image bauen

   docker build -t uae-os-proxy .

2. Container ausführen

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

Umgebungsvariablen

Die Anwendung wird vollständig über Umgebungsvariablen konfiguriert (verwaltet durch app/config.py). Siehe .env.example für Details.

Hinweis: Wenn Supabase-Variablen fehlen, wechselt die App für Testzwecke automatisch in einen Simulationsmodus.

API-Nutzungsbeispiele (cURL)

Der Proxy-Endpunkt (/proxy/execute) ist geschützt und erfordert Ihren API_KEY. Sie können sich entweder mit dem Standard-Header Authorization: Bearer <key> oder dem Header X-API-Key: <key> authentifizieren.

1. Grundlegender Tool-Aufruf (Keine Zahlung)

Mit 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-Tool-Aufruf mit x402-Mikrozahlung

Um eine Abrechnung zu simulieren und an einen anderen Agenten weiterzuleiten, fügen Sie die Felder payment_amount und target_agent_id hinzu.

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
  }'

Erwartete Antwort:

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

Deployment (Kostenlos / Ein-Klick)

Das Dockerfile ist für moderne PaaS-Anbieter optimiert. Es verwendet ein leichtgewichtiges Python 3.11 Slim-Image, exponiert Port 8000 und bindet an 0.0.0.0.

Deployment auf Railway (Empfohlen)

Railway bietet eine nahtlose Ein-Klick-Deployment-Erfahrung, die die enthaltene railway.toml und das Dockerfile automatisch liest.

1. Push zu GitHub: Committen Sie dieses Repository in ein GitHub-Repo.

2. Railway verbinden: Loggen Sie sich bei Railway ein und klicken Sie auf New Project -> Deploy from GitHub repo.

3. Repo auswählen: Wählen Sie Ihr neu gepushtes Repository aus.

4. Umgebungsvariablen setzen: Gehen Sie in Ihrem Railway-Projekt-Dashboard zum Tab Variables und fügen Sie Ihren API_KEY, WEBHOOK_SECRET und alle optionalen Supabase/Stripe-Keys hinzu. Stellen Sie sicher, dass Sie in der Produktion starke zufällige Zeichenfolgen für die Geheimnisse generieren.

5. Deploy: Railway erkennt automatisch die railway.toml-Datei, baut das Docker-Image und stellt es bereit.

Erwartetes Live-URL-Muster: https://agent-economy-os-production.up.railway.app

Produktionsüberwachung & Verifizierung

Sobald das Deployment abgeschlossen ist, können Sie überprüfen, ob der Dienst läuft, und seinen Status über die öffentlichen Endpunkte überwachen. Dies sind die empfohlenen Endpunkte für PaaS-Health-Checks:

• Health Check: https://agent-economy-os-production.up.railway.app/health

  • Gibt grundlegenden Status, Version und Zeitstempel zurück.

• Metriken: https://agent-economy-os-production.up.railway.app/metrics

  • Gibt Betriebszeit und die Gesamtzahl der gemessenen Agenten zurück.

Hinweis: Da das aktuelle Rate Limiting und Caching In-Memory sind, stellen Sie sicher, dass Ihre PaaS so konfiguriert ist, dass sie eine einzelne Instanz/Replik ausführt (was bei kostenlosen Tarifen der Standard ist), bis Redis vollständig integriert ist.

Stripe-Webhook-Einrichtung

Um die Echtzeit-Abrechnungsverfolgung für Live-Stripe-Zahlungen zu aktivieren, konfigurieren Sie einen Webhook in Ihrem Stripe-Dashboard:

1. Gehen Sie im Stripe-Dashboard zu Developers > Webhooks.

2. Klicken Sie auf Add endpoint.

3. Setzen Sie die Endpoint URL auf https://agent-economy-os-production.up.railway.app/webhooks/stripe.

4. Wählen Sie die Events to send: payment_intent.succeeded und payment_intent.payment_failed.

5. Klicken Sie auf Add endpoint.

6. Lassen Sie sich das Signing secret anzeigen (beginnt mit whsec_) und fügen Sie es Ihren Railway-Umgebungsvariablen als STRIPE_WEBHOOK_SECRET hinzu.

Mitwirken

Wir freuen uns über Beiträge! Das Universal Agent Economy OS wurde entwickelt, um der definitive Open-Source-Standard für die agentische Ökonomie zu sein.

Bitte stellen Sie sicher, dass alle Tests bestehen (pytest -v) und die Abdeckung bei 100% bleibt, bevor Sie einen Pull Request einreichen. Wenn Sie ein neues Modul hinzufügen, stellen Sie sicher, dass es auf der bestehenden Architektur aufbaut, ohne den Kern-Proxy-Ausführungsfluss zu unterbrechen.