Unified API: Full Test Example

Below is a Python script that demonstrates how to use the major endpoints. You can adapt it by changing the variables.

PYTHON
Copy
import requests
import json
import websockets
import asyncio
import base64

BASE_URL = "https://api.levangielaboratories.com"
AUTH_TOKEN = "YOUR_USER_ID:TIMESTAMP:RANDOM"
API_KEY = "YOUR_API_KEY"
AGENT_ID = "YOUR_AGENT_ID"

# 1. Check your credits via /info
info_payload = {
    "auth_token": AUTH_TOKEN,
    "action": "get_credits"
}
res = requests.post(f"{BASE_URL}/info", json=info_payload)
print("[INFO] Credits:", res.json())

# 2. Create an API Key via /createapikey
create_key_payload = {
    "auth_token": AUTH_TOKEN,
    "key_name": "MyTestKey"
}
res = requests.post(f"{BASE_URL}/createapikey", json=create_key_payload)
print("[CREATE KEY] Response:", res.json())
new_api_key = res.json().get("key")

# 3. Send a message operation
operation_payload = {
    "operation": "message",
    "params": {"message": "Hello from test script"},
    "api_key": API_KEY,
    "agent_id": AGENT_ID
}
res = requests.post(f"{BASE_URL}/operation", json=operation_payload)
print("[OPERATION] Response:", res.json())

# Save WebSocket URLs for streaming
ws_url = res.json().get("ws_url")
persistent_ws_url = res.json().get("persistent_ws_url")

# 4. Upload a file
file_path = "agent_data/test.txt"
with open("./local_test_file.txt", "rb") as f:
    files = {
        "file": f
    }
    data = {
        "api_key": API_KEY,
        "agent_id": AGENT_ID,
        "file_path": file_path
    }
    res = requests.post(f"{BASE_URL}/upload_file", data=data, files=files)
    print("[UPLOAD FILE] Response:", res.json())

# 5. Get agent history
history_payload = {
    "api_key": API_KEY,
    "agent_id": AGENT_ID,
    "history_type": "chat",
    "filter_count": "5"
}
res = requests.post(f"{BASE_URL}/agent_history", json=history_payload)
print("[AGENT HISTORY] Response:", res.json())

# 6. Connect to the WebSocket for streaming
async def stream_from_websocket():
    # Connect to the persistent WebSocket
    async with websockets.connect(persistent_ws_url) as websocket:
        print("[WebSocket] Connected successfully")
        try:
            while True:
                # Receive message
                message = await websocket.recv()
                data = json.loads(message)
                print(f"[WebSocket] Received: {data}")
        except websockets.exceptions.ConnectionClosed:
            print("[WebSocket] Connection closed")

# Run the WebSocket streaming (uncomment to test)
# asyncio.run(stream_from_websocket())

# 7. Stop agent when done
stop_payload = {
    "operation": "stop_agent",
    "params": {
        "message": "Test completed",
        "reason": "End of test script"
    },
    "api_key": API_KEY,
    "agent_id": AGENT_ID
}
res = requests.post(f"{BASE_URL}/operation", json=stop_payload)
print("[STOP AGENT] Response:", res.json())

# 8. Check health status
res = requests.get(f"{BASE_URL}/health")
print("[HEALTH] Status:", res.json())

Here's a more detailed example of WebSocket streaming with both connection types:

PYTHON
Copy
import asyncio
import websockets
import json

# Session-specific WebSocket (for a single operation)
async def stream_session(ws_url):
    async with websockets.connect(ws_url) as websocket:
        print("Connected to session-specific WebSocket")
        try:
            while True:
                message = await websocket.recv()
                data = json.loads(message)
                print(f"Session: {data}")
        except websockets.exceptions.ConnectionClosed:
            print("Session WebSocket connection closed")

# Persistent WebSocket (for all agent activity)
async def stream_persistent(persistent_ws_url):
    async with websockets.connect(persistent_ws_url) as websocket:
        print("Connected to persistent WebSocket")
        try:
            while True:
                message = await websocket.recv()
                data = json.loads(message)
                print(f"Persistent: {data}")
        except websockets.exceptions.ConnectionClosed:
            print("Persistent WebSocket connection closed")

# Use both streams concurrently
async def stream_both(ws_url, persistent_ws_url):
    # Create tasks for both streams
    session_task = asyncio.create_task(stream_session(ws_url))
    persistent_task = asyncio.create_task(stream_persistent(persistent_ws_url))
    
    # Wait for both to complete
    await asyncio.gather(session_task, persistent_task)

# Run with your WebSocket URLs from an operation response
# asyncio.run(stream_both(ws_url, persistent_ws_url))

The API returns appropriate HTTP status codes for different error scenarios:

• 400 Bad Request: Invalid parameters (e.g., missing required fields, invalid agent ID format)
• 401 Unauthorized: Invalid API key or auth token
• 402 Payment Required: Insufficient credits
• 403 Forbidden: Not authorized to access the requested resource
• 404 Not Found: Agent or resource not found
• 409 Conflict: Agent in wrong status for the requested operation
• 429 Too Many Requests: Rate limit exceeded
• 500 Internal Server Error: Server-side error

Always check response status codes and handle errors appropriately in your application.