ckit_bot_exec
RobotContext, event handlers, main loop patterns. View Reference
The Flexus Client Kit (flexus-client-kit) is the Python SDK for building Flexus bots. It handles communication with the backend, tool registration, event handling, and integrations.
pip install flexus-client-kitckit_bot_exec
RobotContext, event handlers, main loop patterns. View Reference
ckit_cloudtool
Tool definitions, WaitForSubchats, NeedsConfirmation. View Reference
ckit_kanban
Kanban board operations, task creation and management. View Reference
ckit_ask_model
Bot activation, subchats, message creation. View Reference
| Module | Purpose |
|---|---|
ckit_bot_exec | Core bot execution, RobotContext, event handlers |
ckit_cloudtool | Tool definitions and call handling |
ckit_kanban | Kanban board operations |
ckit_ask_model | LLM interactions, subchats |
ckit_client | GraphQL client for Flexus API |
ckit_bot_install | Marketplace registration |
ckit_schedule | Schedule definitions |
ckit_shutdown | Graceful shutdown handling |
ckit_mongo | MongoDB credential fetching |
| Module | Purpose |
|---|---|
fi_pdoc | Policy document CRUD |
fi_mongo_store | Personal MongoDB storage |
fi_slack | Slack messenger integration |
fi_discord2 | Discord messenger integration |
fi_gmail | Gmail integration |
fi_report | PDF report generation |
fi_widget | UI widget rendering |
import asynciofrom flexus_client_kit import ckit_client, ckit_bot_exec, ckit_cloudtool, ckit_shutdownfrom mybot import mybot_install
BOT_NAME = "mybot"BOT_VERSION = "1.0.0"
MY_TOOL = ckit_cloudtool.CloudTool( strict=True, name="my_tool", description="Does something", parameters={"type": "object", "properties": {...}, "required": [...], "additionalProperties": False})
TOOLS = [MY_TOOL]
async def main_loop(fclient, rcx): setup = ckit_bot_exec.official_setup_mixing_procedure( mybot_install.mybot_setup_schema, rcx.persona.persona_setup )
@rcx.on_tool_call(MY_TOOL.name) async def handle(toolcall, args): return "Result"
while not ckit_shutdown.shutdown_event.is_set(): await rcx.unpark_collected_events(sleep_if_no_work=10.0)
def main(): scenario_fn = ckit_bot_exec.parse_bot_args() fclient = ckit_client.FlexusClient( ckit_client.bot_service_name(BOT_NAME, BOT_VERSION), endpoint="/v1/jailed-bot" )
asyncio.run(ckit_bot_exec.run_bots_in_this_group( fclient, marketable_name=BOT_NAME, marketable_version_str=BOT_VERSION, bot_main_loop=main_loop, inprocess_tools=TOOLS, scenario_fn=scenario_fn, install_func=mybot_install.install, ))
if __name__ == "__main__": main()| Variable | Required | Description |
|---|---|---|
FLEXUS_API_BASEURL | Yes | Backend URL (e.g., https://flexus.team/) |
FLEXUS_API_KEY | Yes | Your API key |
FLEXUS_WORKSPACE | No | Default workspace ID |