Getting Started
Installation and first steps with the AGNT5 Python SDK
Get up and running with the AGNT5 Python SDK in minutes. This guide covers installation, your first worker, and local development setup.
Installation
System Requirements
- Python 3.8 or higher
- pip or uv package manager
Install from PyPI
pip install agnt5Development Installation
For development or contributing to the SDK:
git clone https://github.com/agnt5/agnt5
cd agnt5/sdk/sdk-python
pip install -e .Verify Installation
import agnt5
print(agnt5.__version__)First Worker
Create a simple worker with a greeting function:
import asyncio
from agnt5 import Worker, function
@function()
def greet(name: str) -> str:
"""Greet a user by name."""
return f"Hello, {name}!"
@function("math_add")
def add_numbers(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
async def main():
worker = Worker(service_name="hello-service")
await worker.run()
if __name__ == "__main__":
asyncio.run(main())Deploy to AGNT5
Install the CLI and Authenticate
# Install the AGNT5 CLI
brew install agnt5/tap/agnt5 # macOS
# or
curl -LsSf https://agnt5.com/cli.sh | bash # Linux
# Authenticate
agnt5 auth login
# Deploy your worker
agnt5 deployRun Your Worker Locally
In a new terminal, run your worker:
python worker.pyYou should see output like:
INFO:agnt5.worker:Starting worker for service: hello-service
INFO:agnt5.worker:Registered function: greet
INFO:agnt5.worker:Registered function: math_add
INFO:agnt5.worker:Worker running, waiting for tasks...Test Your Functions
Using HTTP API
Test your functions using the Gateway HTTP API:
# Test the greet function
curl -X POST http://localhost:8080/call \
-H "Content-Type: application/json" \
-d '{
"serviceName": "hello-service",
"handlerName": "greet",
"inputData": "QWxpY2U="
}'The inputData is base64-encoded JSON. For "Alice", the base64 is "QWxpY2U=".
Using Python Client
import asyncio
import json
import base64
from agnt5 import Client
async def test_functions():
client = Client("http://localhost:8080")
# Test greet function
name = "Alice"
input_data = base64.b64encode(json.dumps(name).encode()).decode()
result = await client.call(
service_name="hello-service",
handler_name="greet",
input_data=input_data
)
print(f"Greeting result: {result}")
# Test add function
numbers = {"a": 5, "b": 3}
input_data = base64.b64encode(json.dumps(numbers).encode()).decode()
result = await client.call(
service_name="hello-service",
handler_name="math_add",
input_data=input_data
)
print(f"Addition result: {result}")
asyncio.run(test_functions())ASGI Integration
For web applications, use the ASGI runtime:
from agnt5 import Worker, function
@function()
def api_handler(request: dict) -> dict:
return {
"message": "Hello from AGNT5!",
"received": request
}
# Create ASGI application
app = Worker("web-service", runtime="asgi")
app.enable_cors() # Enable CORS for browser accessRun with uvicorn:
pip install uvicorn
uvicorn asgi_app:app --reload --port 8000Test the ASGI endpoints:
# Health check
curl http://localhost:8000/health
# List available functions
curl http://localhost:8000/functions
# Call a function
curl -X POST http://localhost:8000/invoke/api_handler \
-H "Content-Type: application/json" \
-d '{"test": "data"}'Configuration
Environment Variables
Configure your worker using environment variables:
export AGNT5_COORDINATOR_ENDPOINT=http://localhost:9091
export AGNT5_SERVICE_NAME=my-service
export AGNT5_LOG_LEVEL=DEBUG
python worker.pyConfiguration in Code
import logging
from agnt5 import Worker
from agnt5.logging import install_opentelemetry_logging
# Configure logging
logging.basicConfig(level=logging.INFO)
install_opentelemetry_logging()
# Create worker with custom configuration
worker = Worker(
service_name="configured-service",
service_version="1.2.0",
coordinator_endpoint="http://localhost:9091",
runtime="standalone"
)Error Handling
Handle errors gracefully in your functions:
from agnt5 import function
import logging
logger = logging.getLogger(__name__)
@function()
def safe_divide(a: float, b: float) -> dict:
try:
if b == 0:
return {"error": "Division by zero", "result": None}
result = a / b
logger.info(f"Division successful: {a} / {b} = {result}")
return {"result": result, "error": None}
except Exception as e:
logger.error(f"Unexpected error: {e}")
return {"error": str(e), "result": None}Development Tips
Hot Reload
During development, restart your worker when code changes:
import asyncio
import sys
from pathlib import Path
from agnt5 import Worker, function
# Add auto-reload during development
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nWorker stopped")
sys.exit(0)Debugging
Enable debug logging to see detailed execution information:
import logging
logging.basicConfig(level=logging.DEBUG)
from agnt5.logging import install_opentelemetry_logging
install_opentelemetry_logging(level=logging.DEBUG)Testing Functions
Test your functions locally without the full platform:
from agnt5.decorators import execute_component
# Test function directly
result = execute_component("greet", b'{"name": "Alice"}')
print(result)Next Steps
Core Primitives
- Functions - Stateless operations with retries
- Entities - Stateful components with unique keys
- Workflows - Multi-step orchestration
- Context API - Full API reference
Agent Development Kit
- Agents - Autonomous LLM-driven systems
- Tools - Extend agent capabilities
- Sessions - Conversation management
- Memory - Long-term knowledge storage
Configuration
- Worker Runtime - Configure and deploy workers
- Examples - Real-world usage patterns