emergence-agent 0.1.0

Official Python SDK for building and deploying intelligent agents on the Emergence Platform

Emergence Agent SDK

Official Python SDK for building and deploying intelligent agents on the Emergence Platform. Create powerful AI agents with automatic platform integration, capability declaration, and seamless communication.

✨ Features

• 🚀 Ultra-Simple Agent Creation - Inherit from Agent class and override methods
• 🔄 Automatic Platform Integration - Registration, heartbeat, and lifecycle management
• 🎯 Declarative Capabilities - Define what your agent can do with structured schemas
• 🌐 Webhook & Async Support - Built-in HTTP servers and async processing
• 🎨 Powerful Decorators - Validation, caching, rate limiting, retry logic, and more
• 🔍 Agent Discovery - Find and communicate with other agents on the platform
• 📊 Built-in Monitoring - Performance stats, logging, and status reporting
• 🛡️ Production Ready - Error handling, graceful shutdown, and robust design

🚀 Quick Start

Installation

pip install emergence-agent

Basic Agent in 30 Seconds

from emergence_agent import Agent

class MyAgent(Agent):
    def setup(self):
        # Declare what your agent can do
        self.declare_capability("greeting", "Say hello to users")
    
    def handle_request(self, request_type, data):
        # Handle incoming requests
        if request_type == "greet":
            return {"message": f"Hello, {data['name']}!"}
        return {"error": "Unknown request"}

# Start your agent (auto-registers with platform)
with MyAgent(name="greeter") as agent:
    print("🤖 Agent is ready!")
    # Agent automatically handles platform communication

Set Your API Key

export EMERGENCE_API_KEY="your_api_key_here"

Or pass it directly:

agent = MyAgent(name="greeter", api_key="your_api_key")

🎨 Powerful Decorators

Enhance your agent methods with built-in decorators:

from emergence_agent import (
    Agent, capability, validate_input, rate_limit, 
    cache_result, retry_on_failure
)

class AdvancedAgent(Agent):
    @capability(
        "text_analysis",
        "Analyze text with sentiment detection",
        parameters={"text": {"type": "string", "required": True}},
        returns={"sentiment": {"type": "string"}}
    )
    @validate_input(text={"type": "string", "required": True, "min_length": 1})
    @rate_limit(requests_per_second=10, burst_size=20)
    @cache_result(ttl_seconds=300)  # Cache for 5 minutes
    @retry_on_failure(max_retries=3, delay=1.0)
    def analyze_text(self, text: str):
        return {"sentiment": self.detect_sentiment(text)}

Available Decorators:

• @capability() - Auto-declare capabilities
• @validate_input() - Input parameter validation
• @rate_limit() - Rate limiting protection
• @retry_on_failure() - Automatic retry with backoff
• @cache_result() - Method result caching
• @timing_stats() - Performance monitoring

📖 Documentation

Core Components

EmergenceClient

Main client for platform API interactions:

from emergence_agent import EmergenceClient

client = EmergenceClient(
    api_key="your_api_key",
    base_url="https://api.emergence-platform.com",
    timeout=30
)

# List available agents
agents = client.list_agents()

# Get agent details
agent_info = client.get_agent("agent_id")

WebhookAgent

Base class for webhook-enabled agents:

from emergence_agent import WebhookAgent

agent = WebhookAgent(
    name="data-processor",
    description="Processes incoming data streams",
    version="1.0.0",
    webhook_url="https://your-domain.com/webhook"
)

@agent.webhook_handler('/data', methods=['POST'])
def handle_data(payload):
    # Process the data
    return {"processed": True, "count": len(payload.get('items', []))}

AsyncAgent

For high-performance async operations:

from emergence_agent import AsyncAgent
import asyncio

agent = AsyncAgent(name="async-processor")

@agent.async_handler('/batch')
async def process_batch(payload):
    # Async processing
    results = await asyncio.gather(
        *[process_item(item) for item in payload['items']]
    )
    return {"results": results}

Configuration

Environment Variables

EMERGENCE_API_KEY=your_api_key_here
EMERGENCE_BASE_URL=https://api.emergence-platform.com
EMERGENCE_WEBHOOK_URL=https://your-agent.com/webhook
EMERGENCE_LOG_LEVEL=INFO

Config File

from emergence_agent import EmergenceConfig

config = EmergenceConfig(
    api_key="your_api_key",
    agent_name="my-agent",
    webhook_url="https://myagent.com/webhook",
    auto_register=True,
    retry_attempts=3
)

agent = WebhookAgent.from_config(config)

Decorators and Utilities

Validation

from emergence_agent.utils import validate_payload

@agent.webhook_handler('/secure')
@validate_payload({
    "type": "object",
    "required": ["user_id", "data"],
    "properties": {
        "user_id": {"type": "string"},
        "data": {"type": "object"}
    }
})
def secure_endpoint(payload):
    return {"user": payload["user_id"], "processed": True}

Rate Limiting

from emergence_agent.utils import rate_limit

@agent.webhook_handler('/limited')
@rate_limit(requests=100, window=3600)  # 100 requests per hour
def limited_endpoint(payload):
    return {"status": "processed"}

Retry Logic

from emergence_agent.utils import retry_on_failure

@agent.webhook_handler('/reliable')
@retry_on_failure(max_attempts=3, backoff=2.0)
def reliable_endpoint(payload):
    # This will retry on failures with exponential backoff
    result = external_api_call(payload)
    return result

🛠️ CLI Tools

The SDK includes command-line tools for agent development:

Validate Agent

emergence-validate my_agent.py

Deploy Agent

emergence-deploy --agent my_agent.py --platform production

Generate Template

emergence-agent init --template webhook --name my-new-agent

📚 Examples

Flask Integration

from flask import Flask, request
from emergence_agent import EmergenceClient

app = Flask(__name__)
client = EmergenceClient(api_key="your_api_key")

@app.route('/webhook/register', methods=['POST'])
def register_webhook():
    # Register webhook with platform
    result = client.register_webhook(request.json)
    return result

@app.route('/webhook/ping', methods=['POST'])
def ping_webhook():
    # Handle platform ping
    return {"status": "active", "timestamp": datetime.utcnow().isoformat()}

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

FastAPI Integration

from fastapi import FastAPI
from emergence_agent import EmergenceClient
from pydantic import BaseModel

app = FastAPI()
client = EmergenceClient(api_key="your_api_key")

class WebhookPayload(BaseModel):
    event: str
    data: dict

@app.post('/api/webhook/register')
async def register_webhook(payload: dict):
    result = await client.async_register_webhook(payload)
    return result

@app.post('/api/webhook/ping')
async def ping_webhook():
    return {"status": "active", "version": "1.0.0"}

Background Tasks with Celery

from celery import Celery
from emergence_agent import EmergenceClient

app = Celery('agent_tasks')
client = EmergenceClient(api_key="your_api_key")

@app.task
def process_webhook_data(payload):
    # Process data in background
    result = heavy_computation(payload)
    
    # Report back to platform
    client.report_result(payload['task_id'], result)
    return result

🧪 Testing

Unit Tests

import pytest
from emergence_agent import WebhookAgent
from emergence_agent.testing import MockEmergenceClient

def test_webhook_handler():
    agent = WebhookAgent(name="test-agent")
    client = MockEmergenceClient()
    
    @agent.webhook_handler('/test')
    def test_handler(payload):
        return {"echo": payload}
    
    # Test the handler
    result = agent.handle_webhook('/test', {"message": "hello"})
    assert result["echo"]["message"] == "hello"

Integration Tests

import pytest
from emergence_agent import EmergenceClient

@pytest.mark.integration
def test_platform_integration():
    client = EmergenceClient(api_key="test_key")
    
    # Test agent registration
    agent_data = {
        "name": "integration-test-agent",
        "version": "1.0.0",
        "webhook_url": "https://test.example.com/webhook"
    }
    
    result = client.register_agent(agent_data)
    assert result["status"] == "registered"

📦 Installation Options

Basic Installation

pip install emergence-agent

Development Installation

pip install emergence-agent[dev]

Full Installation (with examples)

pip install emergence-agent[all]

From Source

git clone https://github.com/emergence-platform/emergence-agent-sdk.git
cd emergence-agent-sdk
pip install -e .

🔧 Development

Setting up Development Environment

git clone https://github.com/emergence-platform/emergence-agent-sdk.git
cd emergence-agent-sdk
pip install -e .[dev]
pre-commit install

Running Tests

pytest tests/

Code Formatting

black emergence_agent/
isort emergence_agent/
flake8 emergence_agent/

Type Checking

mypy emergence_agent/

🤝 Contributing

We welcome contributions! Please see our Contributing Guide for details.

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Links

• Platform: https://emergence-platform.com
• Documentation: https://docs.emergence-platform.com/agent-sdk
• PyPI Package: https://pypi.org/project/emergence-agent/
• GitHub Repository: https://github.com/emergence-platform/emergence-agent-sdk
• Issue Tracker: https://github.com/emergence-platform/emergence-agent-sdk/issues

💬 Support

• Documentation: docs.emergence-platform.com
• Community Forum: community.emergence-platform.com
• Discord: Join our Discord
• Email: [e-mail]

⭐ Star History

If you find this SDK useful, please consider giving it a star on GitHub!

---

Built with ❤️ by the Emergence Platform Team