Skip to content

Source

get_conversation_id_header(request) async

Get the conversation id from the request header.

Parameters:

Name Type Description Default
request Request

The request object.

 required

Returns:

Name Type Description
conversation_id str | None

The conversation ID.
Source code in mycxo/boxtalk/routes/api/external/route.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def get_conversation_id_header(request: Request) -> uuid.UUID:
    """
    Get the conversation id from the request header.

    Parameters:
        request (Request): The request object.

    Returns:
        conversation_id (str | None): The conversation ID.
    """
    decoded_conversation_id = base64.b64decode(
        request.headers.get("X-Rocketfuel-Conversation-Id", "")
    ).decode("utf-8")
    if decoded_conversation_id:
        return uuid.UUID(decoded_conversation_id)

get_or_create_conversation_id(request) async

Get or create a conversation ID for use in the LLM

Parameters:

Name Type Description Default
request Request

The request object.

 required

Returns:

Name Type Description
conversation_id UUID

The conversation ID.
Source code in mycxo/boxtalk/routes/api/external/route.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def get_or_create_conversation_id(request: Request) -> Request:
    """
    Get or create a conversation ID for use in the LLM

    Parameters:
        request (Request): The request object.

    Returns:
        conversation_id (uuid.UUID): The conversation ID.
    """
    conversation_id = await get_conversation_id_header(request)
    if conversation_id:
        decoded_conversation_id = conversation_id
        conversation_id = uuid.UUID(decoded_conversation_id)
    else:
        conversation_id = uuid.uuid4()
    request.state.conversation_id = conversation_id
    return request

introspect_api_token(request, credentials=Depends(http_bearer)) async

Validate the API token from the X-RocketFuel-Token header. The token should be base64 encoded.

Parameters:

Name Type Description Default
request Request

The request object.

 required
credentials HTTPAuthorizationCredentials

The HTTP authorization credentials.

 Depends(http_bearer)
x_rocketfuel_token str | None

The RocketFuel token from the X-RocketFuel-Token header.

 required

Returns:

Name Type Description
request Request

The request object with the validated token.
Source code in mycxo/boxtalk/routes/api/external/route.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
async def introspect_api_token(
    request: Request,
    credentials: HTTPAuthorizationCredentials = Depends(http_bearer),
) -> Request:
    """
    Validate the API token from the X-RocketFuel-Token header.
    The token should be base64 encoded.

    Parameters:
        request (Request): The request object.
        credentials (HTTPAuthorizationCredentials): The HTTP authorization credentials.
        x_rocketfuel_token (str | None): The RocketFuel token from the X-RocketFuel-Token header.

    Returns:
        request (Request): The request object with the validated token.
    """

    try:
        # Decode the base64 token
        decoded_token = base64.b64decode(credentials.credentials).decode("utf-8")

        # Validate the API key
        is_valid = await validate_api_key_from_db(decoded_token)

        if not is_valid:
            raise NotAuthorizedException(detail="Invalid API token")

        return request

    except base64.binascii.Error:
        raise HTTPException(
            status_code=400, detail="Invalid token format - must be base64 encoded"
        )
    except Exception as e:
        print(f"Error validating API token: {str(e)}")
        raise HTTPException(status_code=500, detail="Error validating API token")

set_conversation_id_header(response, conversation_id) async

Set the conversation id in the response header.

Parameters:

Name Type Description Default
response Response

The response object.

 required
conversation_id UUID

The conversation ID.

 required
Source code in mycxo/boxtalk/routes/api/external/route.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
async def set_conversation_id_header(
    response: Response | JSONResponse, conversation_id: uuid.UUID
) -> Response:
    """
    Set the conversation id in the response header.

    Parameters:
        response (Response): The response object.
        conversation_id (uuid.UUID): The conversation ID.
    """
    encoded_conversation_id = base64.b64encode(conversation_id.bytes).decode("utf-8")
    response.headers["X-Rocketfuel-Conversation-Id"] = encoded_conversation_id
    return response

RocketfuelResults

Bases: BaseModel

The results of the Rocketfuel agent.

Attributes:

Name Type Description
thoughts list[str]

The thoughts of the agent
sql_query str

The SQL query to be executed
result str

The result of the SQL query in CSV format
Source code in mycxo/boxtalk/models/rocketfuel_results.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
class RocketfuelResults(BaseModel):
    """
    The results of the Rocketfuel agent.

    Attributes:
        thoughts (list[str]): The thoughts of the agent
        sql_query (str): The SQL query to be executed
        result (str): The result of the SQL query in CSV format
    """

    thoughts: list[str] = Field(description="The thoughts of the agent")
    sql_query: str = Field(description="The SQL query to be executed")
    result: str = Field(description="The result of the SQL query in CSV format")