import asyncio
import aiohttp
API_KEY = "SLNG_API_KEY"
BASE_URL = "https://api.agents.slng.ai"
async def dispatch_call_async(session, agent_id, phone_number, arguments=None):
async with session.post(
f"{BASE_URL}/v1/agents/{agent_id}/calls",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
},
json={
"phone_number": phone_number,
"arguments": arguments or {}
}
) as response:
if response.status == 200:
data = await response.json()
return {"phone": phone_number, "success": True, "call_id": data["call_id"]}
else:
error = await response.json()
return {
"phone": phone_number,
"success": False,
"error": error.get("error") or error.get("detail") or str(error),
}
async def dispatch_batch(agent_id, contacts, concurrency=5):
semaphore = asyncio.Semaphore(concurrency)
async def limited_dispatch(session, contact):
async with semaphore:
return await dispatch_call_async(
session, agent_id, contact["phone"], contact.get("args")
)
async with aiohttp.ClientSession() as session:
tasks = [limited_dispatch(session, contact) for contact in contacts]
return await asyncio.gather(*tasks)
# Usage
contacts = [
{"phone": "+14155551234", "args": {"customer_name": "John Smith"}},
{"phone": "+14155555678", "args": {"customer_name": "Jane Doe"}},
{"phone": "+14155559012", "args": {"customer_name": "Bob Wilson"}},
]
results = asyncio.run(dispatch_batch("AGENT_ID", contacts))
for result in results:
if result["success"]:
print(f"✓ {result['phone']}: {result['call_id']}")
else:
print(f"✗ {result['phone']}: {result['error']}")