Composio + Stream Deck

This is a workflow builder that ensures the LLM produces a complete, step-by-step plan for any use case. WHEN TO CALL: - Call this tool based on COMPOSIO_SEARCH_TOOLS output. If search tools response indicates create_plan should be called and the usecase is not easy, call it. - Use this tool after COMPOSIO_SEARCH_TOOLS or COMPOSIO_MANAGE_CONNECTIONS to generate an execution plan for the user's use case. - USE for medium or hard tasks — skip it for easy ones. - If the user switches to a new use case in the same chat and COMPOSIO_SEARCH_TOOLS again instructs you to call the planner, you MUST call this tool again for that new use case. Memory Integration: - You can choose to add the memory received from the search tool into the known_fields parameter of the plan function to enhance planning with discovered relationships and information. Outputs a complete plan with sections such as "workflow_steps", "complexity_assessment", "decision_matrix", "failure_handling" "output_format", and more as needed. If you skip this step for non-easy tasks, workflows will likely be incomplete, or fail during execution for complex tasks. Calling it guarantees reliable, accurate, and end-to-end workflows aligned with the available tools and connections.

Execute a tool using the composio api.

Fast and parallel tool executor for tools discovered through COMPOSIO_SEARCH_TOOLS. Use this tool to execute up to 50 tools in parallel across apps only when they're logically independent (no ordering/output dependencies). Response contains structured outputs ready for immediate analysis - avoid reprocessing them via remote bash/workbench tools. Prerequisites: - Always use valid tool slugs and their arguments. NEVER invent tool slugs or argument fields. ALWAYS pass STRICTLY schema-compliant arguments with each tool execution. - Ensure an ACTIVE connection exists for the toolkits that are going to be executed. If none exists, MUST initiate one via COMPOSIO_MANAGE_CONNECTIONS before execution. - Only batch tools that are logically independent - no ordering, no output-to-input dependencies, and no intra-call chaining (tools in one call can't use each other's outputs). DO NOT pass dummy or placeholder inputs; always resolve required inputs using appropriate tools first. Usage guidelines: - If COMPOSIO_SEARCH_TOOLS returns a tool that can perform the task, prefer calling it via this executor. Do not write custom API calls or ad-hoc scripts for tasks that can be completed by available Composio tools. - Prefer parallel execution: group independent tools into a single multi-execute call where possible. - Predictively set sync_response_to_workbench=true if the response may be large or needed for later scripting. It still shows response inline; if the actual response data turns out small and easy to handle, keep everything inline and SKIP workbench usage. - Responses contain structured outputs for each tool. RULE: Small data - process yourself inline; large data - process in the workbench. - ALWAYS include inline references/links to sources in MARKDOWN format directly next to the relevant text. Eg provide slack thread links alongside with summary, render document links instead of raw IDs. Restrictions: Some tools or toolkits may be disabled in this environment. If the response indicates a restriction, inform the user and STOP execution immediately. Do NOT attempt workarounds or speculative actions. - CRITICAL: You MUST always include the 'memory' parameter - never omit it. Even if you think there's nothing to remember, include an empty object {} for memory. Memory Storage: - CRITICAL FORMAT: Memory must be a dictionary where keys are app names (strings) and values are arrays of strings. NEVER pass nested objects or dictionaries as values. - CORRECT format: {"slack": ["Channel general has ID C1234567"], "gmail": ["John's email is john@example.com"]} - Write memory entries in natural, descriptive language - NOT as key-value pairs. Use full sentences that clearly describe the relationship or information. - ONLY store information that will be valuable for future tool executions - focus on persistent data that saves API calls. - STORE: ID mappings, entity relationships, configs, stable identifiers. - DO NOT STORE: Action descriptions, temporary status updates, logs, or "sent/fetched" confirmations. - Examples of GOOD memory (store these): * "The important channel in Slack has ID C1234567 and is called #general" * "The team's main repository is owned by user 'teamlead' with ID 98765" * "The user prefers markdown docs with professional writing, no emojis" (user_preference) - Examples of BAD memory (DON'T store these): * "Successfully sent email to john@example.com with message hi" * "Fetching emails from last day (Sep 6, 2025) for analysis" - Do not repeat the memories stored or found previously.

Tool Server Info: Composio connects 500+ apps—Slack, GitHub, Notion, Google Workspace (Gmail, Sheets, Drive, Calendar), Microsoft (Outlook, Teams), X/Twitter, Figma, Web Search / Deep research, Browser tool (scrape URLs, browser automation), Meta apps (Instagram, Meta Ads), TikTok, and more—for seamless cross-app automation. Use this tool to discover relevant tools plus the recommended plan and common pitfalls for reliable execution. Always call this tool first whenever a user mentions or implies an external app, service, or workflow—never say "I don't have access to X/Y app" before calling it. Usage guidelines: - Use this tool whenever kicking off a task. Re-run it when you need additional tools/plans due to missing details, errors, or a changed use case. - Use search_strategy: "auto" normally. If the returned plan does not match the current request or an expected tool is missing, retry the same queries with search_strategy: "tool_search" to bypass cached plans and run direct tool search. - If the user pivots to a different use case in same chat, you MUST call this tool again with the new use case and generate a new session_id. - Specify the use_case with a normalized description of the problem, query, or task. Be clear and precise. Queries can be simple single-app actions or multiple linked queries for complex cross-app workflows. - Pass known_fields along with use_case as a string of key–value hints (for example, "channel_name: general") to help the search resolve missing details such as IDs. Splitting guidelines (Important): 1. Atomic queries: 1 query = 1 tool call. Include hidden prerequisites (e.g., add "get Linear issue" before "update Linear issue"). 2. Include app names: If user names a toolkit, include it in every sub query so intent stays scoped (e.g., "fetch Gmail emails", "reply to Gmail email"). 3. English input: Translate non-English prompts while preserving intent and identifiers. Example: User query: "send an email to John welcoming him and create a meeting invite for tomorrow" Search call: queries: [ {use_case: "send an email to someone", known_fields: "recipient_name: John"}, {use_case: "create a meeting invite", known_fields: "meeting_date: tomorrow"} ] Plan review checklist (Important): - The response includes a detailed execution plan and common pitfalls. You MUST review this plan carefully, adapt it to your current context, and generate your own final step-by-step plan before execution. Execute the steps in order to ensure reliable and accurate execution. Skipping or ignoring required steps can lead to unexpected failures. - Check the plan and pitfalls for input parameter nuances (required fields, IDs, formats, limits). Before executing any tool, you MUST review its COMPLETE input schema and provide STRICTLY schema-compliant arguments to avoid invalid-input errors. - Determine whether pagination is needed; if a response returns a pagination token and completeness is implied, paginate until exhaustion and do not return partial results. Response: - Tools & Input Schemas: The response lists toolkits (apps) and tools suitable for the task, along with their tool_slug, description, input schema / schemaRef, and related tools for prerequisites, alternatives, or next steps. - NOTE: Tools with schemaRef instead of input_schema require you to call COMPOSIO_GET_TOOL_SCHEMAS first to load their full input_schema before use. - Connection Info: If a toolkit has an active connection, the response includes it along with any available current user information. If no active connection exists, you MUST initiate a new connection via COMPOSIO_MANAGE_CONNECTIONS with the correct toolkit name. DO NOT execute any toolkit tool without an ACTIVE connection. - Time Info: The response includes the current UTC time for reference. You can reference UTC time from the response if needed. - The tools returned to you through this are to be called via COMPOSIO_MULTI_EXECUTE_TOOL. Ensure each tool execution specifies the correct tool_slug and arguments exactly as defined by the tool's input schema. - The response includes a memory parameter containing relevant information about the use case and the known fields that can be used to determine the flow of execution. Any user preferences in memory must be adhered to. SESSION: ALWAYS set this parameter, first for any workflow. Pass session: {generate_id: true} for new workflows OR session: {id: "EXISTING_ID"} to continue. ALWAYS use the returned session_id in ALL subsequent meta tool calls.

Deprecated: use check active connections instead for bulk operations. check active connection status for a toolkit or specific connected account id. returns connection details if active, or required parameters for establishing connection if none exists. active connections enable agent actions on the toolkit.

Check active connection status for multiple toolkits or specific connected account ids. returns connection details if active, or required parameters for establishing connection if none exists. active connections enable agent actions on toolkits.

Enable a specific trigger for the authenticated user.

Process REMOTE FILES or script BULK TOOL EXECUTIONS using Python code IN A REMOTE SANDBOX. If you can see the data in chat, DON'T USE THIS TOOL. ONLY use this when processing data stored in a remote file or when scripting bulk tool executions. DO NOT USE - When the complete response is already inline/in-memory, or you only need quick parsing, summarization, or basic math. USE IF - To parse/analyze tool outputs saved by COMPOSIO_MULTI_EXECUTE_TOOL to a remote file in the sandbox or to script multi-tool chains there. - For bulk or repeated executions of known Composio tools (e.g., add a label to 100 emails). - To call APIs via proxy_execute when no Composio tool exists for that API. OUTPUTS - Returns a compact result or, if too long, artifacts under /mnt/files/.composio/output (cloud-backed FUSE mount, persisted across sandbox restarts). IMPORTANT CODING RULES: 1. Stepwise Execution: Split work into small steps. Save intermediate outputs to /mnt/files/ (cloud-backed, persisted across failures/timeouts) or variables. Call COMPOSIO_REMOTE_WORKBENCH again for the next step. 2. Notebook Persistence: This is a persistent Jupyter notebook cell: variables, functions, imports, and in-memory state persist across executions. Helper functions are preloaded. 3. Top-level cells: Do not use return; Jupyter only allows it inside functions. For final values, end with output or print(output), not return output. 4. Parallelism & Timeout (CRITICAL): There is a hard 3-minute (180s) execution limit per cell. Always prioritize PARALLEL execution using ThreadPoolExecutor for bulk operations - e.g., call run_composio_tool or invoke_llm across rows. If data is large, split it into smaller batches across cells. 5. Checkpoints: Save checkpoints to /mnt/files/ so that long runs can be resumed from the last completed step, even after a timeout or sandbox restart. 6. Schema Safety: Never assume the response schema for run_composio_tool if not known already from previous tools. To inspect schema, either run a simple request outside the workbench or use invoke_llm helper. 7. LLM Helpers: Always use invoke_llm helper for summary, analysis, or field extraction on results; prefer it for much better results over ad hoc filtering. 8. Avoid Meta Loops: Do not use run_composio_tool to call COMPOSIO_* meta tools. Only use it for app tools. 9. Pagination: Use when data spans multiple pages. Continue fetching pages with the returned next_page_token or cursor until none remains. Parallelize page fetches when the tool supports page_number. 10. No Hardcoding: Never hardcode data. Load it from files or tool responses, iterating to construct intermediate or final inputs/outputs. 11. If the final output is in a workbench file, use upload_local_file to download it - never expose the raw workbench file path to the user. Prefer to download useful artifacts after task is complete. ENV & HELPERS: - Home directory: /home/user. - NOTE: Helper functions already initialized in the workbench - DO NOT import or redeclare them: - run_composio_tool(tool_slug: str, arguments: dict) -> tuple[Dict[str, Any], str]: Execute a known Composio app tool. Do not invent names; match the tool input schema. Use for loops/parallel/bulk calls. i) run_composio_tool returns JSON with top-level "data". Parse carefully—structure may be nested. - invoke_llm(query: str) -> tuple[str, str]: Invoke an LLM for semantic tasks. Pass MAX 200k characters. i) NOTE Prompting guidance: When building prompts for invoke_llm, prefer f-strings (or concatenation) so literal braces stay intact. If using str.format, escape braces by doubling them ({{ }}). ii) Define the exact JSON schema you want and batch items into smaller groups to stay within token limit. - upload_local_file(*file_paths) -> tuple[Dict[str, Any], str]: Upload sandbox files to Composio S3/R2 storage for user-downloadable artifacts. - proxy_execute(method, endpoint, toolkit, query_params=None, body=None, headers=None) -> tuple[Any, str]: Call a toolkit API directly when no Composio tool exists. Only one toolkit can be invoked with proxy_execute per workbench call - web_search(query: str) -> tuple[str, str]: Search the web for information. - smart_file_extract(sandbox_file_path: str, show_preview: bool = True) -> tuple[str, str]: Extracts text from files in the sandbox (e.g., PDF, image). All helper functions return a tuple (result, error). Always check error before using result. ## Python Helper Functions for LLM Scripting ### run_composio_tool Executes a known Composio tool via backend API. Do NOT call COMPOSIO_* meta tools to avoid cycles. def run_composio_tool(tool_slug: str, arguments: Dict[str, Any]) -> tuple[Dict[str, Any], str] # Returns: (tool_response_dict, error_message) # Success: ({"data": {actual_data}}, "") - Note the top-level data # Error: ({}, "error_message") or (response_data, "error_message") result, error = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 1, "user_id": "me"}) if error: print("GMAIL_FETCH_EMAILS error:", error) else: email_data = result.get("data", {}) print("Fetched:", email_data) ### invoke_llm Calls LLM for reasoning, analysis, and semantic tasks. Pass MAX 200k characters. # Returns: (llm_response, error_message) # Example: analyze tool response with LLM tool_resp, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5, "user_id": "me"}) if not err: parsed = tool_resp.get("data", {}) resp, err2 = invoke_llm(f"Summarize these emails: {parsed}") if not err2: print(resp) # TIP: batch prompts to reduce LLM calls. ### upload_local_file Uploads sandbox files to Composio S3/R2 storage for upload/download requests involving generated sandbox artifacts. Single files upload directly; multiple files are auto-zipped. # Returns: (result_dict, error_string) # Success: ({"s3_url": str, "uploaded_file": str, "type": str, "id": str, "s3key": str, "message": str}, "") # Error: ({}, "error_message") # Single file result, error = upload_local_file("/path/to/report.pdf") # Multiple files are auto-zipped result, error = upload_local_file("/home/user/doc1.txt", "/home/user/doc2.txt") if not error: print("Uploaded:", result["s3_url"]) ### proxy_execute Direct API call to a connected toolkit service. def proxy_execute( method: Literal["GET","POST","PUT","DELETE","PATCH"], endpoint: str, toolkit: str, query_params: Optional[Dict[str, str]] = None, body: Optional[object] = None, headers: Optional[Dict[str, str]] = None, ) -> tuple[Any, str] # Returns: (response_data, error_message) # Example: GET request with query parameters query_params = {"q": "is:unread", "maxResults": "10"} data, error = proxy_execute("GET", "/gmail/v1/users/me/messages", "gmail", query_params=query_params) if not error: print("Success:", data) ### web_search Searches the web via Exa AI. # Returns: (search_results_text, error_message) results, error = web_search("latest developments in AI") if not error: print("Results:", results) ## Best Practices ### Error-first pattern and Defensive parsing (print keys while narrowing) res, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5}) if err: print("error:", err) elif isinstance(res, dict): print("res keys:", list(res.keys())) data = res.get("data") or {} print("data keys:", list(data.keys())) msgs = data.get("messages") or [] print("messages count:", len(msgs)) for m in msgs: print("subject:", m.get("subject", "<missing>")) ### Parallelize within the 3-minute cell timeout Adjust concurrency so all tasks finish within 3 minutes. import concurrent.futures MAX_CONCURRENCY = 10 # Adjust as needed def process_one(item): result, error = run_composio_tool("GMAIL_SEND_EMAIL", item) if error: return {"status": "failed", "error": error} return {"status": "ok", "data": result} with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as ex: results = list(ex.map(process_one, items))