ckit_cloudtool

The ckit_cloudtool module provides classes for defining tools and handling special control flow cases like subchats and human confirmation.

CloudTool

Define tools that your bot can use:

from flexus_client_kit import ckit_cloudtool

MY_TOOL = ckit_cloudtool.CloudTool(
    strict=True,
    name="my_tool",
    description="Describe what this tool does",
    parameters={
        "type": "object",
        "properties": {
            "required_param": {
                "type": "string",
                "description": "A required parameter",
            },
            "optional_param": {
                "type": ["string", "null"],
                "description": "An optional parameter",
            },
            "enum_param": {
                "type": "string",
                "enum": ["option1", "option2", "option3"],
                "description": "Must be one of the options",
            },
        },
        "required": ["required_param", "optional_param", "enum_param"],
        "additionalProperties": False,
    },
)

Constructor Parameters

Parameter	Type	Description
`name`	`str`	Tool name (used in `@rcx.on_tool_call`)
`description`	`str`	Description shown to LLM
`parameters`	`dict`	JSON Schema for parameters
`strict`	`bool`	Enable OpenAI strict mode (recommended)

Strict Mode Requirements

When strict=True (recommended for OpenAI compatibility):

All parameters must be in required — even optional ones
Use ["type", "null"] for optional — not omitting from required
Must have additionalProperties: False — no extra fields allowed

# Correct: optional param with null type
"optional_field": {
    "type": ["string", "null"],  # Can be string or null
    "description": "Optional field",
}
# And include in required:
"required": ["optional_field"]

# Wrong: omitting from required
"required": []  # Doesn't work in strict mode

Methods

openai_style_tool()

Convert to OpenAI function format:

tool_dict = MY_TOOL.openai_style_tool()
# Returns: {"type": "function", "function": {...}}

# For install script:
tools_json = json.dumps([t.openai_style_tool() for t in TOOLS])

WaitForSubchats

Raise this exception to spawn subchats and wait for results:

from flexus_client_kit import ckit_cloudtool, ckit_ask_model

@rcx.on_tool_call("complex_task")
async def handle(toolcall, args):
    # Create subchats
    subchats = await ckit_ask_model.bot_subchat_create_multiple(
        client=fclient,
        who_is_asking="task_handler",
        persona_id=rcx.persona.persona_id,
        first_question=["Analyze A", "Analyze B"],
        first_calls=["null", "null"],
        title=["Analysis A", "Analysis B"],
        fcall_id=toolcall.fcall_id,
        fexp_name="analyst",
    )

    # This pauses the tool call until subchats complete
    raise ckit_cloudtool.WaitForSubchats(subchats)

When subchats complete:

Their results are collected automatically
Combined result becomes the tool call result
Parent conversation continues

Subchat Requirements

Subchats must complete in 10 minutes. The subchat expert’s Lark kernel must set subchat_result to indicate completion.

NeedsConfirmation

Raise this exception to request human approval before executing:

@rcx.on_tool_call("dangerous_action")
async def handle(toolcall, args):
    # Check if already confirmed
    if not toolcall.confirmed_by_human:
        raise ckit_cloudtool.NeedsConfirmation(
            confirm_setup_key="allow_dangerous",      # Setup field for auto-approve
            confirm_command="delete_all_data",        # What action needs approval
            confirm_explanation="This will permanently delete all data. Are you sure?",
        )

    # User confirmed, proceed
    result = delete_all_data()
    return f"Deleted: {result}"

Parameters

Parameter	Type	Description
`confirm_setup_key`	`str`	Setup field that can auto-approve (optional)
`confirm_command`	`str`	Command/action identifier
`confirm_explanation`	`str`	Message shown to user

Confirmation Flow

Handler raises NeedsConfirmation
UI shows confirmation dialog to user
If user confirms, handler is called again with toolcall.confirmed_by_human=True
If user denies, tool call fails with explanation

Auto-Approval

If confirm_setup_key is set and the user has enabled that setup option:

# In setup schema:
"allow_dangerous": {
    "bs_type": "bool",
    "bs_default": False,
    "bs_description": "Auto-approve dangerous actions",
}

# In handler:
raise ckit_cloudtool.NeedsConfirmation(
    confirm_setup_key="allow_dangerous",  # If True, skips confirmation
    ...
)

FCloudtoolCall

Data class received by tool handlers:

@rcx.on_tool_call("my_tool")
async def handle(toolcall: ckit_cloudtool.FCloudtoolCall, args: dict):
    # toolcall fields:
    toolcall.fcall_id           # Unique call ID
    toolcall.fcall_ft_id        # Thread ID
    toolcall.fcall_name         # Tool name
    toolcall.fcall_arguments    # Original JSON string
    toolcall.ws_id              # Workspace ID
    toolcall.connected_persona_id  # Bot's persona ID
    toolcall.confirmed_by_human    # True if user confirmed

Handler Return Values

Tool handlers can return:

String (Most Common)

@rcx.on_tool_call("my_tool")
async def handle(toolcall, args):
    return "Operation completed successfully"

For images, files, or rich content:

@rcx.on_tool_call("generate_image")
async def handle(toolcall, args):
    image_b64 = generate_image(args["prompt"])
    return [
        {"m_type": "text/plain", "m_content": "Here's your image:"},
        {"m_type": "image/png", "m_content": image_b64},
    ]

Error Handling

Return error messages, don’t raise exceptions:

@rcx.on_tool_call("my_tool")
async def handle(toolcall, args):
    try:
        result = external_api_call(args["input"])
        return f"Success: {result}"
    except ExternalAPIError as e:
        # Return error as string so LLM sees it
        return f"Error: {e.message}"
    # Don't let exceptions bubble up - they become generic "Tool error, see logs"

Exception Behavior

Unhandled exceptions in tool handlers are logged but converted to generic “Tool error, see logs” message. The LLM won’t see the actual error. Always catch expected errors and return descriptive messages.

cloudtool_post_result()

Manually post a tool result (rarely needed):

from flexus_client_kit import ckit_cloudtool

await ckit_cloudtool.cloudtool_post_result(
    client=fclient,
    fcall_id=toolcall.fcall_id,
    result_content="Manual result",
    result_provenance={"custom": "data"},
)

Normally, returning from a handler automatically posts the result. Use this only for advanced cases.

Complete Example

from flexus_client_kit import ckit_cloudtool, ckit_ask_model

# Tool that might need confirmation or spawn subchats
ANALYZE_TOOL = ckit_cloudtool.CloudTool(
    strict=True,
    name="analyze",
    description="Analyze data with optional deep analysis",
    parameters={
        "type": "object",
        "properties": {
            "data": {"type": "string", "description": "Data to analyze"},
            "deep": {"type": "boolean", "description": "Perform deep analysis"},
            "destructive": {"type": "boolean", "description": "Allow destructive ops"},
        },
        "required": ["data", "deep", "destructive"],
        "additionalProperties": False,
    },
)


@rcx.on_tool_call(ANALYZE_TOOL.name)
async def handle_analyze(toolcall, args):
    data = args["data"]
    deep = args["deep"]
    destructive = args["destructive"]

    # Require confirmation for destructive operations
    if destructive and not toolcall.confirmed_by_human:
        raise ckit_cloudtool.NeedsConfirmation(
            confirm_setup_key="allow_destructive",
            confirm_command="destructive_analysis",
            confirm_explanation=f"This will modify the original data: {data[:50]}...",
        )

    # Spawn subchats for deep analysis
    if deep:
        subchats = await ckit_ask_model.bot_subchat_create_multiple(
            client=fclient,
            who_is_asking="deep_analyzer",
            persona_id=rcx.persona.persona_id,
            first_question=[
                f"Analyze statistical properties: {data}",
                f"Analyze patterns: {data}",
                f"Analyze anomalies: {data}",
            ],
            first_calls=["null", "null", "null"],
            title=["Stats", "Patterns", "Anomalies"],
            fcall_id=toolcall.fcall_id,
            fexp_name="analyst",
        )
        raise ckit_cloudtool.WaitForSubchats(subchats)

    # Simple analysis
    result = quick_analyze(data)
    return f"Analysis result: {result}"