captivate-ai-api 0.6.3

An API for Captivate conversation and state management

Captivate AI & LLM API

Overview

This API is developed by CaptivateChat to handle its API formats.This flexible messaging and metadata management system built using Pydantic models, designed to handle complex communication scenarios with robust type checking and validation.

Key Components

Models

• ChatRequest: Simple request model for API endpoints
• Captivate: Primary model managing conversation state and responses
• CaptivateResponseModel: Handles response messages and metadata
• ActionModel: Manages actions with flexible payload handling
• ChannelMetadataModel: Stores dynamic channel and conversation metadata

Features

• Dynamic metadata handling
• Immutable session and chat properties
• Flexible message type support
• Custom metadata manipulation
• Conversation title management
• File handling with text extraction and storage management
• Router mode with protected methods
• Agent escalation and routing capabilities

You can install through:

pip install captivate-ai-api

Captivate Payload

Here's the JSON payload you will send in the POST request:

{
    "session_id": "lance_catcher_test_69c35e3e-7ff4-484e-8e36-792a62567b79",
    "endpoint": "action",
    "user_input": "hi man",
    "files": [
        {
            "filename": "document.pdf",
            "type": "application/pdf",
            "file": {},
            "textContent": {
                "type": "file_content",
                "text": "Extracted text content from the PDF...",
                "metadata": {
                    "source": "file_attachment",
                    "originalFileName": "document.pdf",
                    "storageType": "direct"
                }
            },
            "storage": {
                "fileKey": "uploads/1704067200000-abc123-document.pdf",
                "presignedUrl": "https://s3.amazonaws.com/bucket/uploads/1704067200000-abc123-document.pdf?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=...",
                "expiresIn": 1704070800,
                "fileSize": 1024000,
                "processingTime": 15
            }
        }
    ],
    "incoming_action": [
        {
            "id": "sendEmail",
            "payload": {
                "email": "[e-mail]",
                "message": "You are fired"
            }
        }
    ],
    "metadata": {
        "internal": {
            "channelMetadata": {
                "course_id": "abc",
                "channelMetadata": {
                    "channel": "custom-channel",
                    "channelData": {}
                },
                "user": {
                    "firstName": "Lance",
                    "lastName": "safa",
                    "email": "[e-mail]"
                },
                "phoneNumber": null,
                "custom": {
                    "mode": "non-dbfred",
                    "title": {
                        "type": "title",
                        "title": "\"Latest Updates on EU Regulations\""
                    }
                }
            }
        }
    },
    "hasLivechat": false
}

API Structure

The API now has a cleaner structure with a dedicated ChatRequest model:

1. ChatRequest: Simple request model for API endpoints
2. Captivate: Handles both request processing and response management

This makes the API more intuitive and easier to use.

Usage Example

Using the new structure:

from captivate_ai_api import ChatRequest, Captivate, TextMessageModel

# Create a chat request
chat_request = ChatRequest(
    session_id="test_session_123",
    user_input="Hello, how can you help me?",
    metadata={
        "internal": {
            "channelMetadata": {
                "channelMetadata": {"channel": "web"},
                "custom": {"mode": "assistant"}
            }
        }
    },
    hasLivechat=False
)

# Create Captivate instance using factory method
captivate = Captivate.create(chat_request)

# Set response messages
captivate.set_response([
    TextMessageModel(text="Hello! I'm here to help you.")
])

# Get the response
response = captivate.get_response()

Factory Method Usage:

The create() factory method supports multiple input types:

from captivate_ai_api import Captivate, TextMessageModel

# From ChatRequest (recommended)
captivate = Captivate.create(chat_request)

# From dictionary
data = {"session_id": "123", "user_input": "Hello", ...}
captivate = Captivate.create(data)

# Backward compatibility - this still works:
captivate = Captivate(**chat_request.model_dump())     # Direct constructor

File Handling

The Python library receives files from the frontend (JavaScript SDK) in a comprehensive format and processes them accordingly.

File Structure (Received from Frontend)

The frontend sends files with this structure:

{
    "filename": "document.pdf",
    "type": "application/pdf",
    "file": {},
    "textContent": {
        "type": "file_content",
        "text": "Extracted text content from the PDF...",
        "metadata": {
            "source": "file_attachment",
            "originalFileName": "document.pdf",
            "storageType": "direct"
        }
    },
    "storage": {
        "fileKey": "uploads/1704067200000-abc123-document.pdf",
        "presignedUrl": "https://s3.amazonaws.com/bucket/uploads/1704067200000-abc123-document.pdf?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=...",
        "expiresIn": 1704070800,
        "fileSize": 1024000,
        "processingTime": 15
    }
}

Note: This is the format sent by the Captivate Chat API JavaScript SDK. The Python library receives and processes this complete structure.

File Processing Example

from captivate_ai_api import ChatRequest, Captivate, TextMessageModel

# Create a chat request with files (as received from frontend)
chat_request = ChatRequest(
    session_id="file_session_123",
    user_input="Please analyze these documents",
    files=[
        {
            "filename": "report.pdf",
            "type": "application/pdf",
            "file": {},
            "textContent": {
                "type": "file_content",
                "text": "Q1 2024 Financial Report\n\nRevenue: $1.2M\nExpenses: $800K\nNet Profit: $400K",
                "metadata": {
                    "source": "file_attachment",
                    "originalFileName": "report.pdf",
                    "storageType": "direct"
                }
            },
            "storage": {
                "fileKey": "uploads/1704067200000-abc123-report.pdf",
                "presignedUrl": "https://s3.amazonaws.com/bucket/uploads/1704067200000-abc123-report.pdf?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=...",
                "expiresIn": 1704070800,
                "fileSize": 2048000,
                "processingTime": 25
            }
        }
    ],
    metadata={
        "internal": {
            "channelMetadata": {
                "channelMetadata": {"channel": "web"},
                "custom": {"mode": "assistant"}
            }
        }
    },
    hasLivechat=False
)

# Create Captivate instance
captivate = Captivate.create(chat_request)

# Access file information
files = captivate.get_files()
for file in files:
    print(f"File: {file.get('filename')}")
    print(f"Type: {file.get('type')}")
    print(f"Size: {file.get('storage', {}).get('fileSize')} bytes")
    print(f"Text: {file.get('textContent', {}).get('text', '')[:100]}...")

# Set response with file analysis
captivate.set_response([
    TextMessageModel(text=f"I've analyzed {len(files)} file(s) and found relevant information.")
])

# Get the response
response = captivate.get_response()

File Access Methods

# Get all files
files = captivate.get_files()

# Access individual file properties
for file in files:
    filename = file.get('filename')
    file_type = file.get('type')
    file_size = file.get('storage', {}).get('fileSize')
    text_content = file.get('textContent', {}).get('text')
    storage_url = file.get('storage', {}).get('presignedUrl')
    
    print(f"Processing {filename} ({file_type}, {file_size} bytes)")
    if text_content:
        print(f"Content preview: {text_content[:100]}...")
    if storage_url:
        print(f"Storage URL: {storage_url}")

Download Files to Memory

# Download a file to memory for processing
file_info = files[0]  # Get first file
file_stream = await captivate.download_file_to_memory(file_info)

# Process the file stream
content = file_stream.read()
print(f"Downloaded {len(content)} bytes")

Legacy usage (still supported):

@app.post("/chat") async def handle_chat(data: any): try: # Create Captivate instance using the request data captivate = Captivate(**data.dict()) captivate.set_conversation_title('Lord of the rings')

    # Prepare messages
    response_messages = [
        TextMessageModel(text="Welcome to our platform!"),
        ButtonMessageModel(buttons={"title": "Learn More", "options": [{"label":"Yes","value":"Yes"}]}),
        TableMessageModel(table="<table><tr><th>Name</th><th>Age</th></tr><tr><td>Alice</td><td>30</td></tr></table>"),
        CardCollectionModel(cards=[CardMessageModel(
            text="Special Offer",
            description="Get 20% off your next purchase.",
            image_url="https://example.com/offer.png",
            link="https://example.com/deals"
        )]),
        HtmlMessageModel(html="<h2>Today's Highlights</h2><ul><li>News Item 1</li><li>News Item 2</li></ul>"),
        FileCollectionModel(files=[FileModel(type='application/pdf',url="https://example.com/manual.pdf", filename="UserManual.pdf")] ),
        {"type": "custom", "content": "This is a custom message."}
    ]
    
    # Set the response messages
    captivate.set_response(response_messages)

    # Outgoing actions Both 'payload' & 'data' works for backwards compatibliity. Moving forward it is recommended to use 'data'
    outgoing_actions = [
        ActionModel(id="navigate", payload={"url": "https://example.com"}),
        ActionModel(id="submit", data={"form_id": "1234"})
    ] 
    captivate.set_outgoing_action(outgoing_actions)

    return captivate.get_response() #Returns data to captivate platform in the correct format

# Expected Response from `/chat` Endpoint

When you send the POST request to the `/chat` endpoint, the response will look as follows:

```json
{
    "response": [
        {
            "type": "text",
            "text": "Welcome to our platform!"
        },
        {
            "type": "button",
            "buttons": {
                "title": "Learn More",
                "options": {
                    "label":"Yes",
                    "value":"Yes"
                }
            }
        },
        {
            "type": "table",
            "table": "<table><tr><th>Name</th><th>Age</th></tr><tr><td>Alice</td><td>30</td></tr></table>"
        },
        {
            "type": "cards",
            "text": "Special Offer",
            "description": "Get 20% off your next purchase.",
            "image_url": "https://example.com/offer.png",
            "link": "https://example.com/deals"
        },
        {
            "type": "html",
            "html": "<h2>Today's Highlights</h2><ul><li>News Item 1</li><li>News Item 2</li></ul>"
        },
        { 
        "type":"files",
        "title":"these are the files",
        "files":[{
            "type": "application/pdf",
            "url": "https://example.com/manual.pdf",
            "filename": "UserManual.pdf"
            }]
        },
        {
            "type": "alert",
            "RootModel": {
                "priority": "high",
                "message": "System maintenance scheduled."
            }
        }
    ],
    "session_id": "lance_catcher_test_69c35e3e-7ff4-484e-8e36-792a62567b79",
    "metadata": {
        "internal": {
            "channelMetadata": {
                "user": {
                    "firstName": "Lance",
                    "lastName": "safa",
                    "email": "[e-mail]"
                },
                "channelMetadata": {
                    "channel": "custom-channel",
                    "channelData": {}
                },
                "custom": {
                    "mode": "non-dbfred",
                    "title": {
                        "type": "title",
                        "title": "Lord of the rings"
                    }
                },
                "conversationCreatedAt": null,
                "conversationUpdatedAt": null
            }
        }
    },
    "outgoing_action": [
        {
            "id": "navigate",
            "payload": {
                "url": "https://example.com"
            },
            "data": {
                "url": "https://example.com"
            }
        },
        {
            "id": "submit",
            "payload": {
                "form_id": "1234"
            },
            "data": {
                "form_id": "1234"
            }
        }
    ],
    "hasLivechat": false
}

Functions Overview

1. create (Factory Method)

@classmethod
def create(cls, data: Union[ChatRequest, Dict[str, Any]]) -> "Captivate":

• Description: Factory method to create a Captivate instance from various input types.
• Parameters:
  • data: Either a ChatRequest instance or a dictionary containing the data
• Returns: Captivate - A new Captivate instance
• Examples:

# From ChatRequest
chat_request = ChatRequest(session_id="123", ...)
captivate = Captivate.create(chat_request)

# From dictionary
data = {"session_id": "123", "user_input": "Hello", ...}
captivate = Captivate.create(data)

2. get_session_id

def get_session_id(self) -> str:

• Description: Returns the value of session_id.
• Example:

session_id = captivate_instance.get_session_id()

3. get_user_input

def get_user_input(self) -> Optional[str]:

• Description: Returns the value of user_input.
• Example:

user_input = captivate_instance.get_user_input()

4. get_files

def get_files(self) -> Optional[List[Dict[str, Any]]]:

• Description: Returns the list of files attached to the conversation with complete file information including text content and storage details.
• Returns: Optional[List[Dict[str, Any]]] - List of file objects with comprehensive metadata from the frontend
• Example:

files = captivate_instance.get_files()
if files:
    for file in files:
        filename = file.get('filename')
        file_type = file.get('type')
        file_size = file.get('storage', {}).get('fileSize')
        text_content = file.get('textContent', {}).get('text')
        print(f"File: {filename} ({file_type}, {file_size} bytes)")
        if text_content:
            print(f"Content: {text_content[:100]}...")

5. set_conversation_title

def set_conversation_title(self, title: str):

• Description: Sets the conversation title in the custom metadata.
• Example:

captivate_instance.set_conversation_title("New Conversation Title")

6. get_conversation_title

def get_conversation_title(self) -> Optional[str]:

• Description: Retrieves the conversation title from the custom metadata.
• Example:

conversation_title = captivate_instance.get_conversation_title()

7. set_metadata

def set_metadata(self, key: str, value: Any):

• Description: Sets a key-value pair in the custom metadata.
• Example:

captivate_instance.set_metadata("custom_key", "custom_value")

8. get_metadata

def get_metadata(self, key: str) -> Optional[Any]:

• Description: Retrieves the value for a given key in the custom metadata.
• Example:

metadata_value = captivate_instance.get_metadata("custom_key")

9. remove_metadata

def remove_metadata(self, key: str) -> bool:

• Description: Removes a key from the custom metadata.
• Example:

captivate_instance.remove_metadata("custom_key")

10. get_channel

def get_channel(self) -> Optional[str]:

• Description: Retrieves the channel from the metadata.
• Example:

channel = captivate_instance.get_channel()

11. get_user

def get_user(self) -> Optional[UserModel]:

• Description: Retrieves the user from the metadata.
• Example:

user = captivate_instance.get_user()

12. set_user

def set_user(self, user: UserModel) -> None:

• Description: Sets the user in the metadata.
• Example:

captivate_instance.set_user(UserModel(firstName="John", lastName="Doe"))

13. get_created_at

def get_created_at(self) -> Optional[str]:

• Description: Returns the conversationCreatedAt timestamp from the metadata.
• Example:

created_at = captivate_instance.get_created_at()

14. get_updated_at

def get_updated_at(self) -> Optional[str]:

• Description: Returns the conversationUpdatedAt timestamp from the metadata.
• Example:

updated_at = captivate_instance.get_updated_at()

15. get_has_livechat

def get_has_livechat(self) -> bool:

• Description: Returns the value of hasLivechat.
• Example:

has_livechat = captivate_instance.get_has_livechat()

16. set_response

def set_response(self, response: List[Union[TextMessageModel, FileCollectionModel, ButtonMessageModel, TableMessageModel, CardCollectionModel, HtmlMessageModel, dict]]) -> None:

• Description: Sets the response messages in the Captivate instance.
• Example:

captivate_instance.set_response([
            TextMessageModel(text="Welcome to our platform!"),
            ButtonMessageModel(buttons={"title": "Learn More", "action": "navigate"}),
            TableMessageModel(table="<table><tr><th>Name</th><th>Age</th></tr><tr><td>Alice</td><td>30</td></tr></table>"),
            CardCollectionModel(cards=[CardMessageModel(
                text="Special Offer",
                description="Get 20% off your next purchase.",
                image_url="https://example.com/offer.png",
                link="https://example.com/deals"
            )]),
            HtmlMessageModel(html="<h2>Today's Highlights</h2><ul><li>News Item 1</li><li>News Item 2</li></ul>"),
           FileCollectionModel(title="See files below", files=[FileModel(type='application/pdf',url="https://example.com/manual.pdf", filename="UserManual.pdf")] ),
            {"type": "custom", "content": "This is a custom message."}
            ])

17. get_incoming_action

def get_incoming_action(self) -> Optional[List[ActionModel]]:

• Description: Retrieves the incoming actions from the response object, if present.
• Example:

incoming_actions = captivate_instance.get_incoming_action()

18. set_outgoing_action

def set_outgoing_action(self, actions: List[ActionModel]) -> None:

• Description: Sets the outgoing actions in the response object.
• Example:

captivate_instance.set_outgoing_action([
    ActionModel(id="navigate", data={"url": "https://example.com"})
])

19. get_response

def get_response(self) -> Optional[str]:

• Description: Returns the CaptivateResponseModel as a JSON string if it exists, otherwise returns None.
• Example:

response_json = captivate_instance.get_response()

Note: The model_dump() method still works on both ChatRequest and Captivate instances for backward compatibility:

# Both of these work identically:
captivate = Captivate.create(chat_request)                    # Factory method (recommended)
captivate = Captivate(**chat_request.model_dump())           # Direct constructor (backward compatibility)

19. async_send_message

async def async_send_message(self, environment: str = "dev") -> Dict[str, Any]:

• Description: The async_send_message method is an asynchronous function that sends the conversation data (including messages and actions) to the captivate async messsage API endpoint, depending on the environment (dev or prod)
• Example:

# Create an instance of Captivate
captivate = Captivate(session_id="12345", hasLivechat=True, metadata=metadata)

# Set a message and actions
captivate.set_response([TextMessageModel(text="Hello, World!")])

# Send the message to the API in 'dev' environment
response = await captivate.async_send_message(environment="dev")

20. download_file_to_memory

 async def download_file_to_memory(self, file_info: Dict[str, Any]) -> io.BytesIO:

• Description: Downloads a file from the given dictionary and stores it in memory.

• Example:

captivate_instance.download_file_to_memory(file_info)

21. escalate_to_human

def escalate_to_human(self) -> None:

• Description: Sets an outgoing action to escalate the conversation to a human agent.
• Example:

captivate_instance.escalate_to_human()

22. escalate_to_agent_router

def escalate_to_agent_router(self, reason: Optional[str] = None, intent: Optional[str] = None, recommended_agents: Optional[List[str]] = None) -> None:

• Description: Sets an outgoing action to escalate the conversation to an agent router with optional payload data.
• Parameters:
  • reason (str, optional): The reason for escalation
  • intent (str, optional): The user's intent
  • recommended_agents (List[str], optional): Array of agent IDs to recommend
• Example:

# Basic escalation without payload
captivate_instance.escalate_to_agent_router()

# Escalation with all parameters
captivate_instance.escalate_to_agent_router(
    reason="Complex billing inquiry",
    intent="resolve_payment_issue", 
    recommended_agents=["agent_123", "agent_456", "agent_789"]
)

23. Router Mode Management

The Captivate class includes router mode functionality that controls access to certain methods using a decorator pattern.

23.1 enable_router_mode

def enable_router_mode(self) -> None:

• Description: Enables router mode which allows access to protected methods.
• Example:

captivate_instance.enable_router_mode()

23.2 disable_router_mode

def disable_router_mode(self) -> None:

• Description: Disables router mode, preventing access to protected methods.
• Example:

captivate_instance.disable_router_mode()

23.3 is_router_mode

def is_router_mode(self) -> bool:

• Description: Checks if router mode is currently enabled.
• Returns: bool - True if router mode is enabled, False otherwise
• Example:

if captivate_instance.is_router_mode():
    print("Router mode is enabled")

24. Protected Methods (Router Mode Required)

The following methods are protected by the @requires_router_mode decorator and can only be accessed when router mode is enabled:

24.1 set_agents

@requires_router_mode
def set_agents(self, agents_list: List[str]) -> None:

• Description: Sets the agents_list in custom metadata. This can only be set once and requires router mode to be enabled.
• Parameters:
  • agents_list (List[str]): List of agent IDs to set as agents_list
• Example:

# Enable router mode first
captivate_instance.enable_router_mode()

# Set agents list
captivate_instance.set_agents(["agent_001", "agent_002", "agent_003"])

24.2 get_outgoing_action

@requires_router_mode
def get_outgoing_action(self) -> Optional[List[ActionModel]]:

• Description: Retrieves the outgoing actions from the response object. Only available when router mode is enabled.
• Returns: Optional[List[ActionModel]] - List of outgoing actions or None
• Example:

# Enable router mode first
captivate_instance.enable_router_mode()

# Get outgoing actions
outgoing_actions = captivate_instance.get_outgoing_action()

24.3 is_escalating_to_agent_router

@requires_router_mode
def is_escalating_to_agent_router(self) -> Optional[Dict[str, Any]]:

• Description: Checks if the outgoing action is escalating to agent router and returns the payload. Only available when router mode is enabled.
• Returns: Optional[Dict[str, Any]] - The payload if escalating to agent router, None otherwise
• Example:

# Enable router mode first
captivate_instance.enable_router_mode()

# Check if escalating to agent router
payload = captivate_instance.is_escalating_to_agent_router()
if payload:
    print(f"Escalating with payload: {payload}")

25. Router Mode Usage Examples

Complete Router Mode Workflow

# 1. Create Captivate instance
captivate_instance = Captivate(**data)

# 2. Router mode is disabled by default
print(captivate_instance.is_router_mode())  # False

# 3. Protected methods will fail when router mode is disabled
try:
    captivate_instance.set_agents(["agent_001"])
except ValueError as e:
    print(e)  # "set_agents is only available when router mode is enabled."

# 4. Enable router mode
captivate_instance.enable_router_mode()
print(captivate_instance.is_router_mode())  # True

# 5. Now protected methods work
captivate_instance.set_agents(["agent_001", "agent_002"])
agents = captivate_instance.get_agents()  # ["agent_001", "agent_002"]

# 6. Set up escalation
captivate_instance.escalate_to_agent_router(
    reason="Technical issue",
    recommended_agents=["agent_001"]
)

# 7. Check escalation status
payload = captivate_instance.is_escalating_to_agent_router()
print(payload)  # {"reason": "Technical issue", "recommended_agents": ["agent_001"]}

# 8. Disable router mode
captivate_instance.disable_router_mode()

# 9. Protected methods fail again
try:
    captivate_instance.get_outgoing_action()
except ValueError as e:
    print(e)  # "get_outgoing_action is only available when router mode is enabled."

Error Handling

# All protected methods throw consistent errors when router mode is disabled
try:
    captivate_instance.set_agents(["agent_001"])
except ValueError as e:
    print(e)  # "set_agents is only available when router mode is enabled."

try:
    captivate_instance.get_outgoing_action()
except ValueError as e:
    print(e)  # "get_outgoing_action is only available when router mode is enabled."

try:
    captivate_instance.is_escalating_to_agent_router()
except ValueError as e:
    print(e)  # "is_escalating_to_agent_router is only available when router mode is enabled."

26. Decorator Pattern Implementation

The router mode functionality uses a decorator pattern for clean, maintainable code:

def requires_router_mode(func):
    """Decorator to ensure router mode is enabled for specific methods."""
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        if not self._router_mode:
            raise ValueError(f"{func.__name__} is only available when router mode is enabled.")
        return func(self, *args, **kwargs)
    return wrapper

Benefits of the decorator pattern:

• DRY Principle: Single decorator handles all router mode checks
• Clean Code: Methods focus on core logic, validation is automatic
• Easy to Extend: Add @requires_router_mode to any new method
• Consistent Errors: Same error format across all protected methods

27. escalate_to_agent

def escalate_to_agent(self, agent_id: str, reason: Optional[str] = None) -> None:

• Description: Sets an outgoing action to force redirect the conversation to a specific agent.
• Parameters:
  • agent_id (str): The ID of the agent to redirect to
  • reason (str, optional): The reason for the force redirection
• Example:

# Force redirect to a specific agent without reason
captivate_instance.escalate_to_agent("agent_123")

# Force redirect to a specific agent with reason
captivate_instance.escalate_to_agent(
    agent_id="billing_specialist_001", 
    reason="User has complex billing inquiry requiring specialist knowledge"
)

28. set_private_metadata and Private Metadata Usage

def set_private_metadata(self, key: str, value: Any) -> None:

• Description:
  Sets a key-value pair in the private metadata section (custom['private']) of the conversation. This data is for internal use and is not exposed to the end user.
  You can retrieve this value using get_metadata(key).

• Example:

captivate_instance = Captivate(**data_action)

# Set a private metadata key-value pair
captivate_instance.set_private_metadata('my_secret', 123)

# Retrieve the private metadata value
print(captivate_instance.get_metadata('my_secret'))  # Output: 123

Reserved Key Protection

Attempting to set reserved keys like "private", "title", or "conversation_title" in metadata will raise an exception:

try:
    captivate_instance.set_metadata('private', 'should fail')
except Exception as e:
    print('Expected error for reserved key:', e)