Observability

import requests
from typing import Optional, Union, Dict, Any, List, bytes

# configuration variables
ELEVENLABS_API_KEY = "your_elevenlabs_api_key"

ELEVENLABS_API_BASE_URL = "https://api.elevenlabs.io/v1"

def download_elevenlabs_call_audio(conversation_id: str, output_file: Optional[str] = None) -> Union[bytes, bool, None]:
    """
    Download call audio recording from ElevenLabs API.
    
    Args:
        conversation_id (str): The ElevenLabs conversation ID
        output_file (str, optional): Path to save the audio file. If None, returns the audio content
                                     without saving to file.
    
    Returns:
        Union[bytes, bool, None]: If output_file is None, returns the audio content as bytes.
                                 If output_file is provided, returns True if successful.
                                 Returns None if the request fails.
    """
    # API endpoint for retrieving conversation audio
    url = f"{ELEVENLABS_API_BASE_URL}/convai/conversations/{conversation_id}/audio"
    
    # Set up headers with API key
    headers = {
        "xi-api-key": ELEVENLABS_API_KEY,
    }
    
    # Make the request to ElevenLabs API
    response = requests.get(url, headers=headers)
    
    # Check if the request was successful
    if response.status_code == 200:
        # If output file is provided, save the file
        if output_file:
            with open(output_file, 'wb') as f:
                f.write(response.content)
            return True
        
        # Otherwise return the audio content
        return response.content
    else:
        print(f"Error: {response.status_code}")
        print(response.text)
        return None

# Example usage:
# conversation_id = "your_conversation_id"
# 
# # Download and save to a file
# download_elevenlabs_call_audio(conversation_id, "call_recording.mp3")
# 
# # Or just get the audio content without saving
# audio_content = download_elevenlabs_call_audio(conversation_id)

import requests
import json
from typing import Optional, Dict, Any, List, Union

# configuration variables
ELEVENLABS_API_KEY = "your_elevenlabs_api_key"
CEKURA_API_KEY = "your_cekura_api_key"
CEKURA_AGENT_ID = "your_cekura_agent_id"


CEKURA_API_BASE_URL = "https://api.cekura.ai"
ELEVENLABS_API_BASE_URL = "https://api.elevenlabs.io/v1"

def send_to_cekura_observe(conversation_id: str, transcript_data: List[Dict[str, Any]], 
                           audio_content: bytes) -> Optional[Dict[str, Any]]:
    """
    Send ElevenLabs call data to Cekura's observability API.
    
    Args:
        conversation_id (str): The ElevenLabs conversation ID
        transcript_data (List[Dict[str, Any]]): The transcript data from ElevenLabs
        audio_content (bytes): The audio content downloaded from ElevenLabs
        
    Returns:
        Optional[Dict[str, Any]]: The response from the Cekura API or None if request failed
    """
    # Cekura observability API endpoint
    url = f"{CEKURA_API_BASE_URL}/observability/v1/observe/"
    
    # Set up headers with Cekura API key
    headers = {
        "X-CEKURA-API-KEY": CEKURA_API_KEY
    }
    
    # Prepare the data payload
    data = {
        "agent": CEKURA_AGENT_ID,
        "call_id": conversation_id,
        "transcript_type": "elevenlabs",
        "transcript_json": json.dumps(transcript_data)
    }
    
    # Prepare the files payload with the audio content
    files = {
        "voice_recording": (f"elevenlabs_recording_{conversation_id}.mp3", audio_content, "audio/mpeg")
    }
    
    # Make the request to Cekura API
    response = requests.post(url, headers=headers, data=data, files=files)
    
    # Check if the request was successful
    if response.status_code == 201:
        return response.json()
    else:
        print(f"Error: {response.status_code}")
        print(response.text)
        return None

# Example usage:
# conversation_id = "your_conversation_id"
#
# # First download the audio from ElevenLabs
# audio_content = download_elevenlabs_call_audio(conversation_id)
#
# # Get the transcript data from ElevenLabs (this would come from your webhook or API call)
# transcript_data = [...]  # Your ElevenLabs transcript data

# # Send everything to Cekura
# if audio_content:
#     result = send_to_cekura_observe(conversation_id, transcript_data, audio_content)
#     print(f"Call log created with ID: {result['id']}")

python script.py <AGENT_ID> <11LABS_API_KEY>

# python script.py <AGENT_ID> <11LABS_API_KEY>

import requests
import json
import argparse

def get_agent_description(agent_id: str, api_key: str) -> list | None:
    api_url = f"https://api.elevenlabs.io/v1/convai/agents/{agent_id}"
    headers = {
        "Content-Type": "application/json",
        "xi-api-key": api_key
    }

    try:
        print(f"Fetching data for agent ID: {agent_id}...")
        response = requests.get(api_url, headers=headers)
        response.raise_for_status()
        data = response.json()
        print("Successfully fetched data.")
    except requests.exceptions.RequestException as e:
        print(f"Error: Failed to fetch data from API. {e}")
        return None

    agent_description = []

    try:
        system_prompt = data['conversation_config']['agent']['prompt']['prompt']
        agent_description.append({
            "node name": "system",
            "node prompt": system_prompt
        })
    except KeyError:
        print("Warning: Could not find the main system prompt.")
        agent_description.append({
            "node name": "system",
            "node prompt": "Prompt not found in API response."
        })

    workflow = data.get('workflow', {})
    if not workflow:
        print("No workflow found for this agent.")
        return agent_description

    nodes = workflow.get('nodes', {})
    edges = workflow.get('edges', {})

    for node_id, node_data in nodes.items():
        if node_data.get('type') == 'start':
            continue

        node_name = node_data.get('label', 'Unnamed Node')
        node_prompt = node_data.get('additional_prompt', '')
        transfer_conditions = []

        if edges:
            for edge_data in edges.values():
                if edge_data.get('target') == node_id:
                    fwd_cond = edge_data.get('forward_condition')
                    if fwd_cond and fwd_cond.get('type') != 'unconditional':
                        condition_text = fwd_cond.get('condition') or fwd_cond.get('label')
                        if condition_text:
                            transfer_conditions.append(condition_text)

                    bwd_cond = edge_data.get('backward_condition')
                    if bwd_cond:
                        condition_text = bwd_cond.get('condition') or bwd_cond.get('label')
                        if condition_text:
                            transfer_conditions.append(condition_text)

        agent_description.append({
            "node name": node_name,
            "node prompt": node_prompt,
            "transfer conditions": transfer_conditions
        })

    return agent_description

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Fetch and format agent data from the Eleven Labs API.")
    parser.add_argument("agent_id", type=str, help="The ID of the Eleven Labs agent to query.")
    parser.add_argument("api_key", type=str, help="Your Eleven Labs API key.")

    args = parser.parse_args()

    description = get_agent_description(args.agent_id, args.api_key)

    if description:
        print("\n--- Generated Agent Description ---")
        print(json.dumps(description, indent=2))
        print("---------------------------------\n")