Skip to main content

Production-ready Python SDK for the Agent Lobbi multi-agent collaboration platform

Project description

Agent Lobbi Python SDK

PyPI version Python Support License: MIT Tests

A production-ready Python SDK for building and managing AI agents in the Agent Lobbi ecosystem. The Agent Lobbi provides a secure, scalable platform for multi-agent collaboration, task delegation, and real-time communication.

🚀 Features

  • 🤖 Agent Management: Create, register, and manage AI agents with ease
  • 📋 Task Delegation: Delegate complex tasks to available agents
  • 🔄 Real-time Communication: WebSocket-based messaging and event handling
  • 🛡️ Security: Built-in authentication, data validation, and secure communication
  • 🔧 Production-Ready: Comprehensive error handling, logging, and monitoring
  • ⚡ High Performance: Async/await support with connection pooling
  • 🧪 Well Tested: Full test coverage with pytest
  • 📚 CLI Tools: Command-line interface for common operations
  • 🔌 Easy Integration: Simple API with extensive examples

📦 Installation

# Install from PyPI
pip install agent-lobbi-sdk

# Install with development dependencies
pip install agent-lobbi-sdk[dev]

# Install with all optional dependencies
pip install agent-lobbi-sdk[dev,docs,test]

🏃 Quick Start

Creating a Basic Agent

import asyncio
from agent_lobbi_sdk import Agent, Capability

async def main():
    # Define agent capabilities
    capabilities = [
        Capability(
            name="translate",
            description="Translates text between languages",
            input_schema={"text": "string", "target_language": "string"},
            output_schema={"translated_text": "string"},
            tags=["nlp", "translation"]
        )
    ]
    
    # Create agent
    agent = Agent(
        api_key="your_api_key_here",
        agent_type="TranslationAgent",
        capabilities=capabilities,
        lobby_url="http://localhost:8092"
    )
    
    # Define message handler
    @agent.on_message
    async def handle_message(message):
        action = message.payload.get("action")
        
        if action == "translate":
            text = message.payload.get("text", "")
            target_lang = message.payload.get("target_language", "en")
            
            # Your translation logic here
            translated = f"[{target_lang.upper()}] {text}"
            
            return {
                "success": True,
                "translated_text": translated,
                "source_language": "auto-detected"
            }
        
        return {"success": False, "error": "Unknown action"}
    
    # Start the agent
    await agent.start()
    print("Agent started! Press Ctrl+C to stop.")
    
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        await agent.stop()
        print("Agent stopped.")

if __name__ == "__main__":
    asyncio.run(main())

Using the High-Level Client

import asyncio
from agent_lobbi_sdk import AgentLobbiClient

async def main():
    async with AgentLobbiClient("your_api_key") as client:
        # Check system health
        health = await client.health_check()
        print(f"System status: {health}")
        
        # List available agents
        agents = await client.list_agents()
        print(f"Available agents: {len(agents)}")
        
        # Delegate a task
        result = await client.delegate_task(
            task_name="Translation Task",
            task_description="Translate text to Spanish",
            required_capabilities=["translate"],
            task_data={
                "action": "translate",
                "text": "Hello, world!",
                "target_language": "es"
            }
        )
        print(f"Task result: {result}")

asyncio.run(main())

Using the CLI

# Check Agent Lobbi health
agent-lobbi --api-key YOUR_API_KEY health

# List all agents
agent-lobbi --api-key YOUR_API_KEY list-agents

# Delegate a task
agent-lobbi --api-key YOUR_API_KEY delegate-task \
    --task-name "Translation" \
    --task-description "Translate text" \
    --capabilities "translate" \
    --task-data '{"action": "translate", "text": "Hello", "target_language": "es"}'

# Run a basic agent
agent-lobbi --api-key YOUR_API_KEY run-agent \
    --agent-type "UtilityAgent" \
    --capabilities "echo,translate"

📖 Documentation

Core Classes

Agent

The main class for creating and managing agents.

from agent_lobbi_sdk import Agent, Capability

agent = Agent(
    api_key="your_api_key",
    agent_type="MyAgent",
    capabilities=[...],
    agent_id="optional_custom_id",
    lobby_url="http://localhost:8092",
    debug=False,
    max_retries=3,
    retry_delay=1.0,
    heartbeat_interval=30.0,
    timeout=30.0
)

Key Methods:

  • await agent.start(): Start the agent and connect to the lobby
  • await agent.stop(): Stop the agent and cleanup resources
  • @agent.on_message: Decorator to register message handlers

AgentLobbiClient

High-level client for Agent Lobbi operations.

from agent_lobbi_sdk import AgentLobbiClient

async with AgentLobbiClient(api_key, lobby_url) as client:
    # Client operations here
    pass

Key Methods:

  • await client.health_check(): Check system health
  • await client.list_agents(): List all registered agents
  • await client.delegate_task(...): Delegate tasks to agents
  • await client.get_task_status(task_id): Get task status

Capability

Represents an agent capability with schema validation.

from agent_lobbi_sdk import Capability

capability = Capability(
    name="translate",
    description="Translates text between languages",
    input_schema={"text": "string", "target_language": "string"},
    output_schema={"translated_text": "string"},
    tags=["nlp", "translation"],
    version="1.0.0"
)

Error Handling

The SDK provides comprehensive error handling:

from agent_lobbi_sdk import (
    ConnectionError,
    AuthenticationError,
    TaskError,
    ConfigurationError
)

try:
    await agent.start()
except ConnectionError as e:
    print(f"Connection failed: {e}")
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
except ConfigurationError as e:
    print(f"Configuration error: {e}")

Logging

The SDK uses Python's standard logging module:

import logging

# Enable debug logging
logging.getLogger('agent_lobbi_sdk').setLevel(logging.DEBUG)

# Or configure globally
logging.basicConfig(level=logging.INFO)

🧪 Testing

Run the test suite:

# Install test dependencies
pip install agent-lobbi-sdk[test]

# Run tests
pytest

# Run with coverage
pytest --cov=python_sdk --cov-report=html

# Run specific test file
pytest tests/test_client.py -v

🔧 Development

Setting up Development Environment

# Clone the repository
git clone https://github.com/agent-lobbi/agent-lobbi.git
cd agent-lobbi/src/sdk/python_sdk

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install in development mode
pip install -e .[dev]

# Install pre-commit hooks
pre-commit install

Running Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=python_sdk

# Run specific test categories
pytest -m "not integration"  # Skip integration tests
pytest tests/test_client.py  # Run specific file

Code Quality

# Format code
black python_sdk tests examples

# Sort imports
isort python_sdk tests examples

# Lint code
flake8 python_sdk tests examples

# Type checking
mypy python_sdk

📚 Examples

Check out the examples/ directory for more comprehensive examples:

  • basic_agent.py: Simple agent with multiple capabilities
  • task_delegation.py: Task delegation and monitoring
  • advanced_agent.py: Advanced agent with error handling
  • multi_agent_workflow.py: Coordinating multiple agents

🤝 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.

🆘 Support

🗺️ Roadmap

  • Enhanced monitoring and analytics
  • Advanced security features
  • Performance optimizations
  • Cloud deployment guides
  • Integration with popular AI frameworks
  • GraphQL API support
  • Real-time dashboard

🙏 Acknowledgments

  • Built with ❤️ by the Agent Lobbi Team
  • Inspired by the need for better multi-agent coordination
  • Thanks to all our contributors and users

Agent Lobbi - Empowering AI agents to work together seamlessly.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

agent_lobbi_sdk-1.0.2.tar.gz (15.3 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

agent_lobbi_sdk-1.0.2-py3-none-any.whl (13.7 kB view details)

Uploaded Python 3

File details

Details for the file agent_lobbi_sdk-1.0.2.tar.gz.

File metadata

  • Download URL: agent_lobbi_sdk-1.0.2.tar.gz
  • Upload date:
  • Size: 15.3 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.13.5

File hashes

Hashes for agent_lobbi_sdk-1.0.2.tar.gz
Algorithm Hash digest
SHA256 6e02cde071dc1d1db14cb01204f7a2a0b5b48826a4b2b792488a0527a276b6cd
MD5 96be58fef9478293383e0193fc95e66e
BLAKE2b-256 4ce5c91e8bcd6f04dd3c324532413b4dc0f1326cb00101411f504fbc9b9a06fe

See more details on using hashes here.

File details

Details for the file agent_lobbi_sdk-1.0.2-py3-none-any.whl.

File metadata

File hashes

Hashes for agent_lobbi_sdk-1.0.2-py3-none-any.whl
Algorithm Hash digest
SHA256 aeb7ef52f69bb8263fac11ba990a431a110efa8c31d0aef4bc635dc6e3407250
MD5 5015fc844ed88ce366868574a18b335a
BLAKE2b-256 174fb87c19b7019f2fe3edc403289cf313373ba4d689317b813fe3c32ce5484b

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page