private-me-superagi 1.0.1

SuperAGI integration for xLink identity-based authentication (autonomous AI agents, 603× faster M2M auth)

Private.Me xLink for SuperAGI (Python)

Identity-based authentication for SuperAGI autonomous agents.

Python bindings for @private.me/superagi. Build SuperAGI multi-agent systems with cryptographic identity instead of API keys. Zero-config setup, enterprise-grade compliance, no cascading failures from token expiration.

Installation

Prerequisites

This package requires both the Node.js module and Python bindings:

# 1. Install Node.js package
npm install @private.me/superagi

# 2. Install Python bindings
pip install private-me-superagi

Requirements:

• Python 3.8+
• Node.js 18+ (for backend)
• npm (for Node.js package installation)

Quick Start

from private_me import superagi_xlink

# Create autonomous agent (generates identity automatically)
agent = await superagi_xlink.create_agent(
    name='DeploymentAgent',
    role='CI/CD Deployment',
    goals=['Build', 'Test', 'Deploy']
)

# Execute autonomous task
result = await agent.execute({
    'action': 'deploy',
    'environment': 'production'
})

if result['ok']:
    print(f"✓ Deployment complete ({result['value']['stepsCompleted']} steps)")
    for entry in result['value']['auditLog']:
        print(f"  Step {entry['step']}: {entry['action']}")

That's it. No API keys, no OAuth, no configuration files.

API Reference

create_agent(name: str, role: str = None, goals: List[str] = None, max_iterations: int = 25, verbose: bool = False) -> SuperAGIAgent

Create a new SuperAGI agent with xLink identity.

Parameters:

• name (str): Agent name (for logging/debugging)
• role (str, optional): Agent role description
• goals (List[str], optional): List of agent goals
• max_iterations (int): Maximum iterations per task (default: 25)
• verbose (bool): Enable verbose logging (default: False)

Returns:

• SuperAGIAgent: Agent instance with cryptographic identity

Raises:

• RuntimeError: If Node.js module not found

Example:

agent = await superagi_xlink.create_agent(
    name='BuildAgent',
    role='CI/CD Build Automation',
    goals=['Clone repo', 'Run tests', 'Build artifacts'],
    max_iterations=50,
    verbose=True
)
print(f"Agent DID: {agent.get_did()}")
# Output: did:key:z6Mkh...

SuperAGIAgent.get_did() -> str

Get agent's decentralized identifier.

Returns:

• str: Agent DID (e.g., did:key:z6Mk...)

Example:

did = agent.get_did()
print(f"Agent Identity: {did}")

SuperAGIAgent.execute(payload: Dict[str, Any]) -> Dict[str, Any]

Execute autonomous task.

Parameters:

• payload (Dict[str, Any]): Task parameters (action-specific)

Returns:

• Dict[str, Any]: Execution result with keys:
  • ok (bool): Success indicator
  • value (Dict): Result data (if ok=True)
    • taskId (str): Unique task identifier
    • stepsCompleted (int): Number of steps executed
    • auditLog (List[Dict]): Execution audit trail
  • error (str): Error message (if ok=False)

Raises:

• RuntimeError: If Node.js backend fails
• ValueError: If payload is invalid

Example:

result = await agent.execute({
    'action': 'deploy',
    'environment': 'production',
    'repository': 'https://github.com/company/app',
    'branch': 'main'
})

if result['ok']:
    task_id = result['value']['taskId']
    print(f"Task ID: {task_id}")
    print(f"Steps: {result['value']['stepsCompleted']}")
else:
    print(f"Error: {result['error']}")

SuperAGIAgent.send(recipient_did: str, payload: Dict[str, Any]) -> Dict[str, Any]

Send signed message to another agent.

Parameters:

• recipient_did (str): Target agent's DID
• payload (Dict[str, Any]): Message payload

Returns:

• Dict[str, Any]: Result with ok (bool) and optional error (str)

Raises:

• RuntimeError: If Node.js backend fails
• ValueError: If DID or payload is invalid

Example:

# Send message to another agent
result = await agent1.send(
    recipient_did=agent2.get_did(),
    payload={
        'task': 'process-data',
        'data': [1, 2, 3, 4, 5],
        'priority': 'high'
    }
)

if result['ok']:
    print("✓ Message sent successfully")
else:
    print(f"✗ Send failed: {result['error']}")

SuperAGIAgent.get_audit_log() -> List[Dict[str, Any]]

Get agent's execution audit log.

Returns:

• List[Dict[str, Any]]: Audit log entries with:
  • step (int): Step number
  • action (str): Action performed
  • timestamp (str): ISO 8601 timestamp
  • status (str): Step status (success/failure)

Example:

log = agent.get_audit_log()
for entry in log:
    print(f"[{entry['timestamp']}] Step {entry['step']}: {entry['action']}")

SuperAGIAgent.export_identity() -> bytes

Export agent identity for persistence.

Returns:

• bytes: PKCS#8 private key (store securely!)

Raises:

• RuntimeError: If export fails

Example:

# Export identity
identity_bytes = await agent.export_identity()
with open('agent-identity.key', 'wb') as f:
    f.write(identity_bytes)

from_identity(pkcs8: bytes, name: str = 'agent', role: str = None, goals: List[str] = None, verbose: bool = False) -> SuperAGIAgent

Create agent from existing identity.

Parameters:

• pkcs8 (bytes): PKCS#8 private key bytes
• name (str): Agent name (default: 'agent')
• role (str, optional): Agent role
• goals (List[str], optional): Agent goals
• verbose (bool): Enable verbose logging (default: False)

Returns:

• SuperAGIAgent: Restored agent with same DID

Raises:

• RuntimeError: If identity import fails
• ValueError: If PKCS#8 data is invalid

Example:

# Restore identity
with open('agent-identity.key', 'rb') as f:
    identity_bytes = f.read()

restored = await superagi_xlink.from_identity(
    pkcs8=identity_bytes,
    name='DeploymentAgent',
    role='CI/CD',
    goals=['Build', 'Test', 'Deploy'],
    verbose=True
)
print(f"Restored DID: {restored.get_did()}")

Usage Examples

Basic Autonomous Agent

from private_me import superagi_xlink

# Create agent
agent = await superagi_xlink.create_agent(
    name='MyAgent',
    role='Automation',
    goals=['Task1', 'Task2', 'Task3']
)

print(f"Agent DID: {agent.get_did()}")

# Execute task
result = await agent.execute({
    'action': 'process',
    'data': 'input-data'
})

if result['ok']:
    print(f"Completed {result['value']['stepsCompleted']} steps")

CI/CD Pipeline Automation

from private_me import superagi_xlink
import os

async def deploy_to_production():
    """Deploy application using SuperAGI agent"""

    # Create deployment agent
    deployer = await superagi_xlink.create_agent(
        name='DeploymentAgent',
        role='CI/CD Deployment',
        goals=['Clone', 'Build', 'Test', 'Deploy'],
        verbose=True
    )

    # Execute deployment
    result = await deployer.execute({
        'action': 'deploy',
        'environment': os.getenv('DEPLOY_ENV', 'staging'),
        'repository': os.getenv('GITHUB_REPOSITORY'),
        'branch': os.getenv('GITHUB_REF_NAME', 'main')
    })

    if result['ok']:
        print(f"✓ Deployment complete")
        print(f"  Task ID: {result['value']['taskId']}")
        print(f"  Steps: {result['value']['stepsCompleted']}")

        # Print audit trail
        for entry in result['value']['auditLog']:
            status_icon = "✓" if entry['status'] == 'success' else "✗"
            print(f"  {status_icon} [{entry['timestamp']}] {entry['action']}")
    else:
        print(f"✗ Deployment failed: {result['error']}")
        return False

    return True

# Usage
if __name__ == "__main__":
    import asyncio
    asyncio.run(deploy_to_production())

Multi-Agent Orchestration

from private_me import superagi_xlink

class DevOpsPipeline:
    """Orchestrate build → test → deploy pipeline with multiple agents"""

    def __init__(self):
        self.build_agent = None
        self.test_agent = None
        self.deploy_agent = None

    async def initialize(self):
        """Create agents for each stage"""
        self.build_agent = await superagi_xlink.create_agent(
            name='BuildAgent',
            role='Build Automation',
            goals=['Clone', 'Compile', 'Package']
        )

        self.test_agent = await superagi_xlink.create_agent(
            name='TestAgent',
            role='Test Automation',
            goals=['Unit tests', 'Integration tests', 'Security scan']
        )

        self.deploy_agent = await superagi_xlink.create_agent(
            name='DeployAgent',
            role='Deployment',
            goals=['Deploy', 'Verify', 'Monitor']
        )

    async def run_pipeline(self, repository: str, branch: str):
        """Execute complete CI/CD pipeline"""

        # Step 1: Build
        print("Step 1: Building...")
        build_result = await self.build_agent.execute({
            'action': 'build',
            'repository': repository,
            'branch': branch
        })

        if not build_result['ok']:
            print(f"✗ Build failed: {build_result['error']}")
            return False

        build_id = build_result['value']['taskId']
        print(f"✓ Build complete (ID: {build_id})")

        # Step 2: Test
        print("\nStep 2: Testing...")
        test_result = await self.test_agent.execute({
            'action': 'test',
            'build_id': build_id
        })

        if not test_result['ok']:
            print(f"✗ Tests failed: {test_result['error']}")
            return False

        print(f"✓ Tests passed ({test_result['value']['stepsCompleted']} checks)")

        # Step 3: Deploy
        print("\nStep 3: Deploying...")
        deploy_result = await self.deploy_agent.execute({
            'action': 'deploy',
            'build_id': build_id,
            'environment': 'production'
        })

        if not deploy_result['ok']:
            print(f"✗ Deployment failed: {deploy_result['error']}")
            return False

        print(f"✓ Deployment complete")
        return True

# Usage
async def main():
    pipeline = DevOpsPipeline()
    await pipeline.initialize()

    success = await pipeline.run_pipeline(
        repository='https://github.com/company/app',
        branch='main'
    )

    if success:
        print("\n🎉 Pipeline completed successfully!")
    else:
        print("\n❌ Pipeline failed")

# Run
import asyncio
asyncio.run(main())

Multi-Agent Communication

from private_me import superagi_xlink

async def agent_to_agent_messaging():
    """Demonstrate secure agent-to-agent communication"""

    # Create two agents
    coordinator = await superagi_xlink.create_agent(
        name='Coordinator',
        role='Task Coordinator'
    )

    worker = await superagi_xlink.create_agent(
        name='Worker',
        role='Task Executor'
    )

    print(f"Coordinator DID: {coordinator.get_did()}")
    print(f"Worker DID: {worker.get_did()}")

    # Coordinator sends task to worker
    result = await coordinator.send(
        recipient_did=worker.get_did(),
        payload={
            'task': 'process-data',
            'data': [1, 2, 3, 4, 5],
            'deadline': '2026-05-02T12:00:00Z'
        }
    )

    if result['ok']:
        print("✓ Task assigned to worker")
    else:
        print(f"✗ Failed to assign task: {result['error']}")

# Run
import asyncio
asyncio.run(agent_to_agent_messaging())

Enterprise Compliance Automation

from private_me import superagi_xlink
from datetime import datetime
from typing import Dict, List

class ComplianceAutomation:
    """Automated compliance verification with audit trails"""

    def __init__(self):
        self.agent = None

    async def initialize(self):
        """Create compliance agent"""
        self.agent = await superagi_xlink.create_agent(
            name='ComplianceAgent',
            role='SOC2/ISO27001 Compliance',
            goals=['Audit systems', 'Verify controls', 'Generate reports'],
            verbose=True
        )

    async def run_soc2_audit(self, domain: str) -> Dict:
        """Run SOC2 compliance audit"""

        result = await self.agent.execute({
            'action': 'audit',
            'auditType': 'SOC2',
            'domain': domain,
            'timestamp': datetime.utcnow().isoformat()
        })

        if not result['ok']:
            return {
                'passed': False,
                'error': result['error']
            }

        # Process audit results
        audit_log = result['value']['auditLog']
        score = self._calculate_compliance_score(audit_log)

        return {
            'passed': score >= 0.95,
            'score': score,
            'audit_log': audit_log,
            'task_id': result['value']['taskId'],
            'timestamp': datetime.utcnow().isoformat()
        }

    def _calculate_compliance_score(self, audit_log: List[Dict]) -> float:
        """Calculate compliance score from audit log"""
        if not audit_log:
            return 0.0

        passed = sum(1 for entry in audit_log if entry.get('status') == 'success')
        total = len(audit_log)
        return passed / total

# Usage
async def main():
    compliance = ComplianceAutomation()
    await compliance.initialize()

    # Run audit
    result = await compliance.run_soc2_audit('production')

    print(f"\nCompliance Audit Results:")
    print(f"  Score: {result['score']:.1%}")
    print(f"  Status: {'✓ PASSED' if result['passed'] else '✗ FAILED'}")
    print(f"  Checks: {len(result.get('audit_log', []))}")
    print(f"  Task ID: {result.get('task_id', 'N/A')}")

# Run
import asyncio
asyncio.run(main())

GitHub Actions Integration

Python Deployment Script (deploy.py):

from private_me import superagi_xlink
import os
import sys

async def deploy():
    """Deploy via SuperAGI agent in GitHub Actions"""

    deployer = await superagi_xlink.create_agent(
        name='GitHubDeployer',
        role='Deployment',
        goals=['Deploy', 'Verify']
    )

    result = await deployer.execute({
        'action': 'deploy',
        'environment': os.getenv('DEPLOY_ENV', 'staging'),
        'repository': os.getenv('GITHUB_REPOSITORY'),
        'branch': os.getenv('GITHUB_REF_NAME'),
        'commit_sha': os.getenv('GITHUB_SHA')
    })

    if result['ok']:
        print(f"✓ Deployment complete")
        print(f"  Task ID: {result['value']['taskId']}")
        return 0
    else:
        print(f"✗ Deployment failed: {result['error']}")
        return 1

if __name__ == "__main__":
    import asyncio
    sys.exit(asyncio.run(deploy()))

GitHub Actions Workflow (.github/workflows/deploy.yml):

name: Deploy with SuperAGI xLink

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Setup Node.js
        uses: actions/setup-node@v3
        with:
          node-version: '18'

      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'

      - name: Install SuperAGI xLink
        run: npm install -g @private.me/superagi

      - name: Install Python dependencies
        run: pip install private-me-superagi

      - name: Run deployment
        run: python deploy.py
        env:
          DEPLOY_ENV: production

Architecture

This package uses a wrapper pattern:

Python App → Python Bindings → Node.js Backend → SuperAGI Runtime

1. Python layer: Provides Pythonic API (create_agent(), execute(), send())
2. Node.js backend: Handles cryptographic operations (Ed25519, AES-256-GCM)
3. SuperAGI runtime: Autonomous agent execution, goal planning, tool execution
4. xLink protocol: Identity-based M2M authentication

Why Node.js backend?

• Ed25519 cryptography requires native performance
• Message signing/verification happens per-message
• Node.js Crypto API provides optimal implementation
• SuperAGI runtime is JavaScript-based

Error Handling

Type Hints

from typing import Dict, Any, Optional, List
from private_me import superagi_xlink

async def execute_with_retry(
    agent: superagi_xlink.SuperAGIAgent,
    payload: Dict[str, Any],
    max_retries: int = 3
) -> Optional[Dict[str, Any]]:
    """Execute task with automatic retry on failure."""

    for attempt in range(max_retries):
        try:
            result = await agent.execute(payload)

            if result['ok']:
                return result['value']
            else:
                print(f"Attempt {attempt + 1} failed: {result['error']}")

        except RuntimeError as e:
            print(f"Runtime error on attempt {attempt + 1}: {e}")

        if attempt < max_retries - 1:
            import asyncio
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

    return None

Custom Error Classes

from private_me import superagi_xlink

class AgentError(Exception):
    """Base exception for agent errors."""
    pass

class ExecutionError(AgentError):
    """Task execution failed."""
    pass

class CommunicationError(AgentError):
    """Agent communication failed."""
    pass

async def safe_execute(
    agent: superagi_xlink.SuperAGIAgent,
    payload: dict
) -> dict:
    """Execute task with comprehensive error handling."""

    try:
        result = await agent.execute(payload)

        if not result['ok']:
            error_msg = result.get('error', 'Unknown error')

            if 'timeout' in error_msg.lower():
                raise ExecutionError(f"Task timed out: {error_msg}")
            elif 'communication' in error_msg.lower():
                raise CommunicationError(f"Communication failed: {error_msg}")
            else:
                raise AgentError(error_msg)

        return result['value']

    except RuntimeError as e:
        raise AgentError(f"Node.js backend error: {e}")
    except Exception as e:
        raise AgentError(f"Unexpected error: {e}")

Validation Example

from private_me import superagi_xlink
from typing import TypedDict

class DeployPayload(TypedDict):
    action: str
    environment: str
    repository: str
    branch: str

async def execute_validated_deploy(
    agent: superagi_xlink.SuperAGIAgent,
    payload: DeployPayload
) -> bool:
    """Execute deployment with payload validation."""

    # Validate payload
    if payload['action'] != 'deploy':
        raise ValueError("action must be 'deploy'")

    if payload['environment'] not in ['staging', 'production']:
        raise ValueError("environment must be 'staging' or 'production'")

    if not payload['repository'].startswith('https://'):
        raise ValueError("repository must be a valid HTTPS URL")

    # Execute validated deployment
    result = await agent.execute(payload)
    return result['ok']

Troubleshooting

"Node.js module not found"

Error:

RuntimeError: Node.js module @private.me/superagi not found

Solution:

# Install Node.js package first
npm install @private.me/superagi

# Verify installation
ls node_modules/@private.me/superagi

"Node.js backend error"

Error:

RuntimeError: Node.js backend error: <message>

Solution:

• Check Node.js version (requires 18+): node --version
• Verify Node.js package installed: npm list @private.me/superagi
• Check agent name is valid (alphanumeric + hyphens only)
• Review Node.js error message for details
• Enable verbose mode: create_agent(..., verbose=True)

Task execution failures

Error:

{'ok': False, 'error': 'Task execution failed'}

Solution:

# Enable verbose logging to see detailed execution steps
agent = await superagi_xlink.create_agent(
    name='agent',
    verbose=True
)

# Check agent goals are well-defined
agent = await superagi_xlink.create_agent(
    name='agent',
    goals=['Specific', 'Measurable', 'Achievable']  # Not vague
)

# Verify payload structure
import json
json.dumps(payload)  # Should not raise exception

Message delivery failures

Error:

{'ok': False, 'error': 'Message delivery failed'}

Solution:

# Verify recipient DID is correct
print(f"Sending to: {recipient.get_did()}")

# Check DID format (must start with 'did:key:')
if not recipient_did.startswith('did:key:'):
    print("Invalid DID format")

# Enable verbose logging
agent = await superagi_xlink.create_agent(name='agent', verbose=True)

Identity import failures

Error:

ValueError: Invalid PKCS#8 data

Solution:

# Verify file contains valid PKCS#8 bytes
with open('agent.key', 'rb') as f:
    data = f.read()
    print(f"Identity file size: {len(data)} bytes")

    # File should be 32-64 bytes for Ed25519 private key
    if len(data) < 32:
        print("File too small - may be corrupted")

# Ensure file was written in binary mode ('wb')
# Text mode ('w') will corrupt the key data

Max iterations exceeded

Error:

{'ok': False, 'error': 'Maximum iterations exceeded'}

Solution:

# Increase max_iterations for complex tasks
agent = await superagi_xlink.create_agent(
    name='agent',
    max_iterations=50  # Default is 25
)

# Or simplify agent goals
agent = await superagi_xlink.create_agent(
    name='agent',
    goals=['Single focused goal']  # Instead of many goals
)

Performance Tips

Agent Pooling

from private_me import superagi_xlink
from typing import Dict

class AgentPool:
    """Pool of reusable SuperAGI agents."""

    def __init__(self):
        self.agents: Dict[str, superagi_xlink.SuperAGIAgent] = {}

    async def get_agent(self, name: str, role: str = None) -> superagi_xlink.SuperAGIAgent:
        """Get or create agent."""
        if name not in self.agents:
            self.agents[name] = await superagi_xlink.create_agent(
                name=name,
                role=role
            )
        return self.agents[name]

# Usage
pool = AgentPool()
deployer = await pool.get_agent('deployer', 'CI/CD')

Batch Operations

import asyncio
from private_me import superagi_xlink

async def batch_execute(
    agent: superagi_xlink.SuperAGIAgent,
    tasks: list[dict]
) -> list[dict]:
    """Execute multiple tasks in parallel."""

    async def execute_single(task):
        return await agent.execute(task)

    # Execute all tasks concurrently
    results = await asyncio.gather(
        *[execute_single(task) for task in tasks],
        return_exceptions=True
    )

    return results

# Execute 10 tasks in parallel
results = await batch_execute(
    agent=worker,
    tasks=[{'action': 'process', 'id': i} for i in range(10)]
)

Development

Running Tests

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest -v

# Run with coverage
pytest -v --cov=private_me --cov-report=html

# Run specific test
pytest tests/test_superagi_agent.py -k test_create_agent

Type Checking

# Install type checker
pip install mypy

# Run type checking
mypy private_me/superagi_xlink

Building

# Build wheel
python setup.py bdist_wheel

# Validate build
bash validate-build.sh

Why SuperAGI Agents Choose xLink

Eliminates cascading failures: OAuth token expires → 500 autonomous agents retry simultaneously → system overload. xLink uses cryptographic identity → no tokens to expire → no cascades possible.

603× faster M2M auth: Identity verification is 91ms vs 54,853ms for OAuth token refresh. Built-in RFC 7807 errors with field-level validation reduce token usage during failure scenarios.

Zero credential management: Identity can't be stolen, never expires, no rotation required. When integration works first time, agents spend fewer tokens on error recovery.

Enterprise compliance: Audit logs built-in, cryptographic identity verification, no secret sprawl in CI/CD pipelines.

Support

• Documentation: https://private.me/docs/superagi
• White Paper: https://private.me/docs/superagi.html
• Email: [e-mail]
• GitHub: https://github.com/xail-io/xail

License

Proprietary - See LICENSE.md

---

Questions? Visit private.me/docs/superagi-xlink for complete documentation.