ElevenLabs Agents

1
import os
2
import json
3
from typing import AsyncGenerator
4
from fastapi import FastAPI, Request, HTTPException, Header
5
from fastapi.responses import StreamingResponse
6
from openai import AsyncOpenAI
7
from zep_cloud.client import AsyncZep
8
9
app = FastAPI()
10
11
# Initialize clients
12
zep = AsyncZep(api_key=os.environ["ZEP_API_KEY"])
13
openai_client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
14
15
# Proxy authentication
16
PROXY_API_KEY = os.environ["PROXY_API_KEY"]
17
18
19
def extract_user_id(request_body: dict) -> str | None:
20
    """Extract user_id from ElevenLabs custom_llm_extra_body."""
21
    extra_body = request_body.get("extra_body", {})
22
    return extra_body.get("user_id")
23
24
25
async def get_zep_context(user_id: str, thread_id: str) -> str:
26
    """Retrieve relevant context from Zep for the current user."""
27
    try:
28
        context = await zep.thread.get_user_context(thread_id=thread_id)
29
        return context.context or ""
30
    except Exception:
31
        return ""
32
33
34
async def ensure_user_and_thread(user_id: str, thread_id: str):
35
    """Create user and thread if they don't exist."""
36
    try:
37
        await zep.user.add(user_id=user_id)
38
    except Exception:
39
        pass  # User already exists
40
41
    try:
42
        await zep.thread.create(thread_id=thread_id, user_id=user_id)
43
    except Exception:
44
        pass  # Thread already exists
45
46
47
async def save_messages_to_zep(thread_id: str, user_message: str, assistant_message: str):
48
    """Persist the conversation turn to Zep."""
49
    from zep_cloud import Message
50
51
    messages = [
52
        Message(role="user", content=user_message),
53
        Message(role="assistant", content=assistant_message),
54
    ]
55
    await zep.thread.add_messages(thread_id=thread_id, messages=messages)

1
@app.post("/v1/chat/completions")
2
async def chat_completions(
3
    request: Request,
4
    authorization: str = Header(None),
5
):
6
    # Validate API key
7
    if not authorization or not authorization.startswith("Bearer "):
8
        raise HTTPException(status_code=401, detail="Missing or invalid authorization")
9
10
    provided_key = authorization.replace("Bearer ", "")
11
    if provided_key != PROXY_API_KEY:
12
        raise HTTPException(status_code=401, detail="Invalid API key")
13
14
    body = await request.json()
15
16
    # Extract user identity from ElevenLabs extra_body
17
    user_id = extract_user_id(body)
18
    if not user_id:
19
        raise HTTPException(status_code=400, detail="user_id required in extra_body")
20
21
    thread_id = f"elevenlabs_{user_id}"
22
23
    # Ensure user and thread exist in Zep
24
    await ensure_user_and_thread(user_id, thread_id)
25
26
    # Get relevant context from Zep
27
    zep_context = await get_zep_context(user_id, thread_id)
28
29
    # Inject context into system prompt
30
    messages = body.get("messages", [])
31
    if messages and messages[0].get("role") == "system":
32
        original_system = messages[0].get("content", "")
33
        messages[0]["content"] = f"{original_system}\n\n## Relevant Context\n{zep_context}"
34
    elif zep_context:
35
        messages.insert(0, {
36
            "role": "system",
37
            "content": f"## Relevant Context\n{zep_context}"
38
        })
39
40
    # Extract the latest user message for persistence
41
    latest_user_message = ""
42
    for msg in reversed(messages):
43
        if msg.get("role") == "user":
44
            latest_user_message = msg.get("content", "")
45
            break
46
47
    # Forward to OpenAI with streaming
48
    model = body.get("model", "gpt-4o-mini")
49
    stream = body.get("stream", True)
50
51
    if stream:
52
        return StreamingResponse(
53
            stream_response(messages, model, thread_id, latest_user_message),
54
            media_type="text/event-stream",
55
        )
56
    else:
57
        response = await openai_client.chat.completions.create(
58
            model=model,
59
            messages=messages,
60
        )
61
62
        # Save conversation to Zep
63
        assistant_message = response.choices[0].message.content or ""
64
        await save_messages_to_zep(thread_id, latest_user_message, assistant_message)
65
66
        return response.model_dump()

1
async def stream_response(
2
    messages: list,
3
    model: str,
4
    thread_id: str,
5
    user_message: str,
6
) -> AsyncGenerator[str, None]:
7
    """Stream the response and persist the complete message to Zep."""
8
    full_response = ""
9
10
    stream = await openai_client.chat.completions.create(
11
        model=model,
12
        messages=messages,
13
        stream=True,
14
    )
15
16
    async for chunk in stream:
17
        # Accumulate the response
18
        if chunk.choices[0].delta.content:
19
            full_response += chunk.choices[0].delta.content
20
21
        # Forward chunk to ElevenLabs
22
        yield f"data: {json.dumps(chunk.model_dump())}\n\n"
23
24
    yield "data: [DONE]\n\n"
25
26
    # Save complete conversation turn to Zep
27
    if user_message and full_response:
28
        await save_messages_to_zep(thread_id, user_message, full_response)

1
if __name__ == "__main__":
2
    import uvicorn
3
    uvicorn.run(app, host="0.0.0.0", port=8000)