Metadata-Version: 2.4
Name: private-me-autogen
Version: 0.2.0
Summary: AutoGen integration for xLink identity-based authentication (multi-agent agentic coordination, 603× faster M2M auth)
Author: Private.Me Contributors
License: SEE LICENSE IN LICENSE.md
Project-URL: Homepage, https://private.me
Project-URL: Documentation, https://private.me/docs/autogen
Project-URL: Repository, https://github.com/private-me/platform
Project-URL: Issues, https://github.com/private-me/platform/issues
Keywords: autogen,xlink,identity,authentication,ai-agents,multi-agent,agentic,coordination,m2m,zero-config
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: Other/Proprietary License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Security :: Cryptography
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: private-me-aci-core>=0.3.1
Requires-Dist: private-me-shared>=0.1.1
Requires-Dist: private-me-ux-helpers>=0.1.1
Requires-Dist: pyautogen>=0.2.0
Requires-Dist: pydantic>=2.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"

# Private.Me xLink for AutoGen (Python)

**Identity-based M2M authentication for AutoGen conversable agents.**

Python bindings for `@private.me/autogen`. Build AutoGen multi-agent systems that communicate securely using Ed25519 DID identity instead of API keys. Eliminates cascading failures caused by token expiration in distributed agent workflows.

## Installation

### Prerequisites

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

```bash
# 1. Install Node.js package
npm install @private.me/autogen autogen-js

# 2. Install Python bindings
pip install private-me-autogen
```

**Requirements:**
- Python 3.9+
- Node.js 18+ (for backend)
- npm (for Node.js package installation)

## Quick Start

```python
from private_me import autogen

# Create agents with automatic identity
researcher = await autogen.create_agent(name='researcher')
writer = await autogen.create_agent(name='writer')

# Agent-to-agent message (no credentials needed!)
result = await researcher.send(
    to=writer.did,
    payload={'task': 'write-summary', 'topic': 'AI trends'}
)

print(result)  # Response from writer agent
```

**That's it.** No API keys, no OAuth, no token refresh logic.

## API Reference

### `create_agent(name: str, verbose: bool = False) -> AutoGenAgent`

Create a new AutoGen agent with xLink identity.

**Parameters:**
- `name` (str): Agent name (for logging/debugging)
- `verbose` (bool): Enable verbose logging (default: False)

**Returns:**
- `AutoGenAgent`: Agent instance with cryptographic identity

**Raises:**
- `RuntimeError`: If Node.js module not found

**Example:**
```python
agent = await autogen.create_agent(
    name='research-agent',
    verbose=True
)
print(f"Agent DID: {agent.did}")
# Output: did:key:z6Mkh...
```

### `AutoGenAgent.send(to: str, payload: Dict[str, Any], scope: str = 'autogen:message') -> Dict[str, Any]`

Send message to another agent.

**Parameters:**
- `to` (str): Target agent DID or agent name
- `payload` (Dict[str, Any]): Message payload
- `scope` (str): Permission scope (default: 'autogen:message')

**Returns:**
- `Dict[str, Any]`: Result indicating success or error

**Raises:**
- `RuntimeError`: If Node.js backend fails
- `ValueError`: If payload is invalid

**Example:**
```python
result = await researcher.send(
    to='did:key:z6Mk...',
    payload={
        'task': 'analyze-data',
        'data': [1, 2, 3, 4, 5]
    },
    scope='autogen:analysis'
)

if result['ok']:
    print("Message sent successfully")
else:
    print(f"Error: {result['error']}")
```

### `AutoGenAgent.receive_message(envelope: Dict[str, Any]) -> Dict[str, Any]`

Receive and decrypt message from another agent.

**Parameters:**
- `envelope` (Dict[str, Any]): Transport envelope from message relay

**Returns:**
- `Dict[str, Any]`: Decrypted message payload

**Raises:**
- `RuntimeError`: If decryption fails
- `ValueError`: If envelope is malformed

**Example:**
```python
# Typically used in message handler
envelope = await relay.fetch_message(agent.did)
result = await agent.receive_message(envelope)

if result['ok']:
    payload = result['value']
    print(f"Received: {payload}")
```

### `AutoGenAgent.export_identity() -> bytes`

Export agent identity for persistence.

**Returns:**
- `bytes`: PKCS#8 private key (store securely!)

**Raises:**
- `RuntimeError`: If export fails

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

### `from_identity(pkcs8: bytes, name: str = 'agent', verbose: bool = False) -> AutoGenAgent`

Create agent from existing identity.

**Parameters:**
- `pkcs8` (bytes): PKCS#8 private key bytes
- `name` (str): Agent name (default: 'agent')
- `verbose` (bool): Enable verbose logging (default: False)

**Returns:**
- `AutoGenAgent`: Restored agent with same DID

**Raises:**
- `RuntimeError`: If identity import fails
- `ValueError`: If PKCS#8 data is invalid

**Example:**
```python
# Restore identity
with open('agent.key', 'rb') as f:
    identity_bytes = f.read()

restored = await autogen.from_identity(
    pkcs8=identity_bytes,
    name='researcher',
    verbose=True
)
print(f"Restored DID: {restored.did}")
```

## AutoGen Integration

### Basic Conversable Agent

```python
from autogen import ConversableAgent
from private_me import autogen

# Create xLink agent
xlink_agent = await autogen.create_agent(name='assistant')

# Create AutoGen conversable agent
assistant = ConversableAgent(
    name="assistant",
    llm_config={"model": "gpt-4"},
    system_message="You are a helpful AI assistant."
)

# Custom message handler with xLink auth
async def handle_secure_message(message):
    # Send to secure service using xLink identity
    result = await xlink_agent.send(
        to='secure-service-did',
        payload={'query': message['content']}
    )
    return result

assistant.register_hook("process_message", handle_secure_message)
```

### Multi-Agent Coordination

```python
from private_me import autogen
from autogen import GroupChat, GroupChatManager

# Create xLink agents for coordination
coordinator = await autogen.create_agent(name='coordinator')
researcher = await autogen.create_agent(name='researcher')
analyst = await autogen.create_agent(name='analyst')
writer = await autogen.create_agent(name='writer')

# Orchestrate workflow
async def research_pipeline(topic: str):
    # Step 1: Coordinator assigns research task
    await coordinator.send(
        to=researcher.did,
        payload={'task': 'research', 'topic': topic}
    )

    # Step 2: Research gathers data
    research_result = await researcher.send(
        to='web-scraper-service',
        payload={'topic': topic, 'max_results': 20}
    )

    # Step 3: Analyst processes data
    await researcher.send(
        to=analyst.did,
        payload={'task': 'analyze', 'data': research_result}
    )

    analysis = await analyst.send(
        to='data-analyzer-service',
        payload={'raw_data': research_result}
    )

    # Step 4: Writer creates summary
    await analyst.send(
        to=writer.did,
        payload={'task': 'summarize', 'analysis': analysis}
    )

    summary = await writer.send(
        to='text-summarizer-service',
        payload={'content': analysis['insights']}
    )

    return summary

# Run pipeline
result = await research_pipeline('AI agent security patterns')
print(result)
```

### Group Chat with xLink Authentication

```python
from autogen import ConversableAgent, GroupChat, GroupChatManager
from private_me import autogen

# Create xLink agents for each participant
manager_xlink = await autogen.create_agent(name='manager')
engineer_xlink = await autogen.create_agent(name='engineer')
tester_xlink = await autogen.create_agent(name='tester')

# Create AutoGen conversable agents
manager = ConversableAgent(name="manager", llm_config={"model": "gpt-4"})
engineer = ConversableAgent(name="engineer", llm_config={"model": "gpt-4"})
tester = ConversableAgent(name="tester", llm_config={"model": "gpt-4"})

# Create group chat
group_chat = GroupChat(
    agents=[manager, engineer, tester],
    messages=[],
    max_round=10
)

# Add xLink-authenticated external service calls
async def secure_code_review(code: str):
    return await engineer_xlink.send(
        to='code-review-service',
        payload={'code': code, 'language': 'python'}
    )

async def secure_test_execution(test_suite: str):
    return await tester_xlink.send(
        to='test-runner-service',
        payload={'suite': test_suite}
    )
```

## Architecture

This package uses a **wrapper pattern**:

```
Python App → Python Bindings → Node.js Backend → xLink Protocol
```

1. **Python layer**: Provides Pythonic API (`create_agent()`, `send()`)
2. **Node.js backend**: Handles cryptographic operations (Ed25519, AES-256-GCM)
3. **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

## Error Handling

### Type Hints

```python
from typing import Dict, Any, Optional
from private_me import autogen

async def send_with_retry(
    agent: autogen.AutoGenAgent,
    to: str,
    payload: Dict[str, Any],
    max_retries: int = 3
) -> Optional[Dict[str, Any]]:
    """Send message with automatic retry on failure."""

    for attempt in range(max_retries):
        try:
            result = await agent.send(to=to, payload=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:
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

    return None
```

### Validation Example

```python
from private_me import autogen
from typing import TypedDict, List

class TaskPayload(TypedDict):
    task: str
    data: List[int]
    priority: int

async def send_validated_task(
    agent: autogen.AutoGenAgent,
    to: str,
    task_data: TaskPayload
) -> bool:
    """Send task with payload validation."""

    # Validate payload
    if not isinstance(task_data['task'], str):
        raise ValueError("task must be a string")

    if not isinstance(task_data['data'], list):
        raise ValueError("data must be a list")

    if task_data['priority'] < 1 or task_data['priority'] > 5:
        raise ValueError("priority must be between 1 and 5")

    # Send validated payload
    result = await agent.send(to=to, payload=task_data)

    return result['ok']
```

### Custom Error Handler

```python
import asyncio
from private_me import autogen

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

class AuthenticationError(AgentError):
    """Authentication failed."""
    pass

class MessageDeliveryError(AgentError):
    """Message delivery failed."""
    pass

async def safe_send(
    agent: autogen.AutoGenAgent,
    to: str,
    payload: dict
) -> dict:
    """Send message with comprehensive error handling."""

    try:
        result = await agent.send(to=to, payload=payload)

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

            # Check error type
            if 'authentication' in error_msg.lower():
                raise AuthenticationError(f"Auth failed: {error_msg}")
            elif 'delivery' in error_msg.lower():
                raise MessageDeliveryError(f"Delivery 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}")
```

## Troubleshooting

### "Node.js module not found"

**Error:**
```
RuntimeError: Node.js module @private.me/autogen not found
```

**Solution:**
```bash
# Install Node.js package first
npm install @private.me/autogen autogen-js

# Verify installation
ls node_modules/@private.me/autogen
```

### "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/autogen`
- Check agent name is valid (alphanumeric + hyphens only)
- Review Node.js error message for details

### Message delivery failures

**Error:**
```
{'ok': False, 'error': 'Message delivery failed'}
```

**Solution:**
```python
# Verify recipient DID is correct
print(f"Sending to: {recipient.did}")

# Check message payload is serializable
import json
json.dumps(payload)  # Should not raise exception

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

### Identity import failures

**Error:**
```
ValueError: Invalid PKCS#8 data
```

**Solution:**
```python
# 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")

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

## Performance Tips

### Connection Pooling

```python
from private_me import autogen
from typing import Dict

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

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

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

# Usage
pool = AgentPool()
agent = await pool.get_agent('researcher')
```

### Batch Operations

```python
import asyncio
from private_me import autogen

async def batch_send(
    agent: autogen.AutoGenAgent,
    recipients: list[str],
    payload: dict
) -> list[dict]:
    """Send message to multiple recipients in parallel."""

    tasks = [
        agent.send(to=recipient, payload=payload)
        for recipient in recipients
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# Send to 10 agents in parallel
results = await batch_send(
    agent=coordinator,
    recipients=[f'agent-{i}' for i in range(10)],
    payload={'task': 'process-batch'}
)
```

## Development

### Running Tests

```bash
# 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_autogen_agent.py -k test_create_agent
```

### Type Checking

```bash
# Install type checker
pip install mypy

# Run type checking
mypy private_me/autogen_xlink
```

### Building

```bash
# Build wheel
python setup.py bdist_wheel

# Validate build
bash validate-build.sh
```

## Why AutoGen Agents Choose xLink

**Eliminates cascading failures:** OAuth token expires → 500 AutoGen 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.

## Support

- **Documentation**: https://private.me/docs/autogen
- **White Paper**: https://private.me/docs/autogen.html
- **Email**: contact@private.me
- **GitHub**: https://github.com/xail-io/xail

## License

Proprietary - See LICENSE.md

---

**Questions?** Visit [private.me/docs/autogen](https://private.me/docs/autogen) for complete documentation.
