class EvalionAPIService:
"""Service class for interacting with the Evalion API."""
def __init__(
self, base_url: str = "https://api.evalion.ai/api/v1", api_token: str = ""
):
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {api_token}"}
async def create_hosted_agent(
self, prompt: str, name: Optional[str] = None
) -> Dict[str, Any]:
"""Create a hosted agent with the given prompt."""
if not name:
name = f"Voice Agent - {uuid.uuid4()}"
payload = {
"name": name,
"description": "Agent created for evaluation",
"agent_type": "outbound",
"prompt": prompt,
"is_active": True,
"speaks_first": False,
"llm_provider": "openai",
"llm_model": "gpt-4o-mini",
"llm_temperature": 0.7,
"tts_provider": "elevenlabs",
"tts_model": "eleven_turbo_v2_5",
"tts_voice": "5IDdqnXnlsZ1FCxoOFYg",
"stt_provider": "openai",
"stt_model": "gpt-4o-mini-transcribe",
"language": "en",
"max_conversation_time_in_minutes": 5,
"llm_max_tokens": 800,
}
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{self.base_url}/hosted-agents",
headers=self.headers,
json=payload,
)
response.raise_for_status()
return response.json()
async def delete_hosted_agent(self, hosted_agent_id: str) -> None:
"""Delete a hosted agent."""
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.delete(
f"{self.base_url}/hosted-agents/{hosted_agent_id}",
headers=self.headers,
)
response.raise_for_status()
async def create_agent(
self,
project_id: str,
hosted_agent_id: str,
prompt: str,
name: Optional[str] = None,
) -> Dict[str, Any]:
"""Create an agent that references a hosted agent."""
if not name:
name = f"Test Agent {int(time.time())}"
payload = {
"name": name,
"description": "Agent for evaluation testing",
"agent_type": "inbound",
"interaction_mode": "voice",
"integration_type": "phone",
"language": "en",
"speaks_first": False,
"prompt": prompt,
"is_active": True,
"hosted_agent_id": hosted_agent_id,
"project_id": project_id,
}
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{self.base_url}/projects/{project_id}/agents",
headers=self.headers,
json=payload,
)
response.raise_for_status()
return response.json()
async def delete_agent(self, project_id: str, agent_id: str) -> None:
"""Delete an agent."""
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.delete(
f"{self.base_url}/projects/{project_id}/agents/{agent_id}",
headers=self.headers,
)
response.raise_for_status()
async def create_test_set(
self, project_id: str, name: Optional[str] = None
) -> Dict[str, Any]:
"""Create a test set."""
if not name:
name = f"Test Set {int(time.time())}"
payload = {
"name": name,
"description": "Test set for evaluation",
"project_id": project_id,
}
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{self.base_url}/projects/{project_id}/test-sets",
headers=self.headers,
json=payload,
)
response.raise_for_status()
return response.json()
async def delete_test_set(self, project_id: str, test_set_id: str) -> None:
"""Delete a test set."""
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.delete(
f"{self.base_url}/projects/{project_id}/test-sets/{test_set_id}",
headers=self.headers,
)
response.raise_for_status()
async def create_test_case(
self, project_id: str, test_set_id: str, scenario: str, expected_outcome: str
) -> Dict[str, Any]:
"""Create a test case."""
payload = {
"name": f"Test Case {int(time.time())}",
"description": "Test case for evaluation",
"scenario": scenario,
"expected_outcome": expected_outcome,
"test_set_id": test_set_id,
}
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{self.base_url}/projects/{project_id}/test-cases",
headers=self.headers,
json=payload,
)
response.raise_for_status()
return response.json()
async def create_test_setup(
self,
project_id: str,
agent_id: str,
persona_id: str,
test_set_id: str,
metrics: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""Create a test setup."""
payload = {
"name": f"Test Setup {int(time.time())}",
"description": "Test setup for evaluation",
"project_id": project_id,
"agents": [agent_id],
"personas": [persona_id],
"test_sets": [test_set_id],
"metrics": metrics or [],
"testing_mode": "voice",
}
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{self.base_url}/test-setups",
headers=self.headers,
json=payload,
)
response.raise_for_status()
return response.json()
async def delete_test_setup(self, project_id: str, test_setup_id: str) -> None:
"""Delete a test setup."""
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.delete(
f"{self.base_url}/test-setups/{test_setup_id}?project_id={project_id}",
headers=self.headers,
)
response.raise_for_status()
async def run_test_setup(self, project_id: str, test_setup_id: str) -> str:
"""Prepare and run a test setup."""
# First prepare
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{self.base_url}/test-setup-runs/prepare",
headers=self.headers,
json={"project_id": project_id, "test_setup_id": test_setup_id},
)
response.raise_for_status()
test_setup_run_id = response.json()["test_setup_run_id"]
# Then run
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.post(
f"{self.base_url}/test-setup-runs/{test_setup_run_id}/run",
headers=self.headers,
json={"project_id": project_id},
)
response.raise_for_status()
return test_setup_run_id
async def poll_for_completion(
self, project_id: str, test_setup_run_id: str, max_wait: int = 600
) -> Optional[Dict[str, Any]]:
"""Poll for simulation completion."""
start_time = time.time()
while time.time() - start_time < max_wait:
async with httpx.AsyncClient(timeout=300.0) as client:
response = await client.get(
f"{self.base_url}/test-setup-runs/{test_setup_run_id}/simulations",
headers=self.headers,
params={"project_id": project_id},
)
if response.status_code == 200:
data = response.json()
simulations = data.get("data", [])
if simulations:
sim = simulations[0]
status = sim.get("run_status")
if status in ["COMPLETED", "FAILED"]:
return sim
await asyncio.sleep(5)
return None
