Composio

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.

This is a workflow builder that ensures the LLM produces a complete, step-by-step plan for any use case.
You MUST always call this tool after COMPOSIO_SEARCH_TOOLS or COMPOSIO_MANAGE_CONNECTIONS to get a proper execution plan for the user's use case.
If the user pivots to a different use case in same chat, you MUST call this tool again with the 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_handlig" "output_format", and more as needed.
If you skip this step, 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.
SESSION: ALWAYS pass session_id from previous meta tool calls. The session_id was in the COMPOSIO_SEARCH_TOOLS response. Required for workflow correlation.
    

    Convert the executed workflow into a notebook.
    
    This tool allows you to:
    1. Save the full workflow execution as a reusable notebook
    2. If I share the notebook with others (when published) it can be used by passing correct environment variables and it executes the whole workflow from beginning of this session
    3. You should generate input json schema of this notebook based on the executed workflow, so that other users of this published notebook can pass their own valid inputs
    4. You should generate the code of this notebook based on the executed workflow, please see below for more instructions on how to generate the code for the notebook
    5. Similarly please generate good output json schema for this notebook, so that other users of this published notebook know how to consume the response
    6. The notebook should take the input from environment variables using os.environ.get(). The user of this notebook will pass each key of the input json schema as an env var
    WHEN TO USE
    - Only run this tool when the workflow is completed and successful or if the user explicitly asked to run this tool
    DO NOT USE
    - When the workflow is still being processed, or not yet completed and the user explicitly didn't ask to run this tool
    IMPORTANT CODING RULES:
    1. Single Execution: Please generate the code for the full notebook that can be executed in a single invocation
    2. 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 via COMPOSIO_MULTI_EXECUTE_TOOL or use invoke_llm helper.
    3. Parallelism & Timeout (CRITICAL): There is a hard timeout of 4 minutes so complete the code within that. Prioritize PARALLEL execution using ThreadPoolExecutor with suitable concurrency for bulk operations - e.g., call run_composio_tool or invoke_llm parallelly across rows to maximize efficiency.
    4. LLM Helpers: You should always use invoke_llm helper for summary, analysis, or field extraction on results. This is a smart LLM that will give much better results than any adhoc filtering.
    5. Avoid Meta Loops: Do not use run_composio_tool to call COMPOSIO_MULTI_EXECUTE_TOOL or other COMPOSIO_* meta tools to avoid cycles. Only use it for app tools.
    6. Pagination: Use when data spans multiple pages. Continue fetching pages with the returned next_page_token or cursor until none remains. Parallelize fetching pages if tool supports page_number.
    7. No Hardcoding: Never hardcode data in code. Always load it from files or tool responses, iterating to construct intermediate or final inputs/outputs.
    8. Code Correctness (CRITICAL): Code must be syntactically and semantically correct and executable.
    9. Please ensure that your code takes input in the form of input json schema via environment variables and stores the output at last command in the form of output json schema
    10. Your notebook must read each key of the input json schema in your code from environment variables using os.environ.get()
    11. Please always end the code with just "output" command following the output json schema so that the notebook actually shows it. DO NOT PRINT IT AT ANY COST
    12. Please only take needed args as input in input json schema and only give the needed data in the output. This is to keep the json schemas and notebook simple
    13. Debugging (CRITICAL): For every print statement in your code, please prefix it with the time at which it is being executed. This will help me investigate latency related stuff
    ENV & HELPERS:
    1. Home directory: `/home/user`.
    2. Helper functions initialized in the sandbox where the notebook is going to run:
      1) `run_composio_tool(tool_slug: str, arguments: dict)`: Execute a known Composio **app** tool (from COMPOSIO_SEARCH_TOOLS). Do not invent names; match the tool’s input schema. Suited for loops/parallel/bulk over datasets.
        1.1) run_composio_tools returns JSON with top-level "data". Parse carefully—structure may be nested.
      2) `invoke_llm(query: str)`: Invoke an LLM for semantic tasks. Pass MAX 400000 characters in input.
      3) `proxy_execute(method, endpoint, toolkit, params=None, body=None)`: Call a toolkit API directly when no Composio tool exists.
      4) `web_search(query: str)`: Search the web for information.
      5) `upload_local_file(*file_paths)`: Upload files to Composio S3/R2 storage. Use this to download any generated files/artifacts from the sandbox.
      6) `smart_file_extract(file_path: str)`: Extract text from any file format (pdf, image, doc, etc). Use this to extract content from workbench files in any format for summary/analysis. Use invoke_llm on response if it's large. Returns Tuple[str, str] (response, error)
    
## Python Helper Functions for LLM Scripting
### 1) run_composio_tool(tool_slug, arguments)
Executes a known Composio tool via backend API. Do NOT call COMPOSIO_* meta tools to avoid cyclic calls.
    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); return
    email_data = result.get("data", {})
    print("Fetched:", email_data)
### 2) invoke_llm(query)
Calls Groq LLM for reasoning, analysis, and semantic tasks. Pass MAX 400000 characters input.
    def invoke_llm(query: str) -> tuple[str, str]
    # Returns: (llm_response, error_message)
    resp, error = invoke_llm("Summarize the key points from this data")
    if error:
        print("invoke_llm error:", error); return
    print("LLM:", resp)
    # Example: analyze tool response with LLM
    tool_resp, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5, "user_id": "me"})
    if err:
        print("GMAIL_FETCH_EMAILS error:", err); return
    parsed = tool_resp.get("data", {})
    resp, err = invoke_llm(f"Analyze these emails and summarize: {parsed}")
    if err:
        print("invoke_llm error:", err); return
    print("LLM Gmail Summary:", resp)
    # TIP: batch prompts to reduce LLM calls.
### 4) proxy_execute(method, endpoint, toolkit, params=None, body=None)
Direct API call to a connected toolkit service.
    from typing import Literal, Optional
    def proxy_execute(
        method: Literal["GET","POST","PUT","DELETE","PATCH"],
        endpoint: str,
        toolkit: str,
        params: Optional[list[Parameter]] = None,
        body: Optional[object] = None,
    ) -> tuple[ToolProxyResponse | None, str]
    # Returns: (response_object, error_message)
    response, error = proxy_execute("GET", "https://api.github.com/repos/owner/repo", "github")
    if error:
        print("proxy_execute error:", error); return
    print("Success:", response.data)
### 5) web_search(query)
Searches the web via Exa AI.
    def web_search(query: str) -> tuple[str, str]
    # Returns: (search_results_text, error_message)
    results, error = web_search("latest developments in AI")
    if error:
        print("web_search error:", error)
    else:
        print("Results:", results)
## Best Practices
### Error-first pattern
    data, error = some_helper(...)
    if error:
        print("some_helper error:", error); return
    # safe to use `data`
### Defensive parsing (print keys while narrowing)
    res, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5})
    if not err and 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 (4-min sandbox timeout)
Adjust concurrency so all tasks finish within 4 minutes.
    import concurrent.futures
    MAX_CONCURRENCY = 10 # Adjust as needed
    def send_bulk_emails(email_list):
        def send_single(email):
            result, error = run_composio_tool("GMAIL_SEND_EMAIL", {
                "to": email["recipient"], "subject": email["subject"], "body": email["body"]
            })
            if error:
                print(f"Failed {email['recipient']}: {error}")
                return {"status": "failed", "error": error}
            return {"status": "sent", "data": result}
        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as ex:
            futures = [ex.submit(send_single, e) for e in email_list]
            for f in concurrent.futures.as_completed(futures):
                results.append(f.result())
        return results
    # Example usage
    email_list = [{"recipient": f"user{i}@example.com", "subject": "Test", "body": "Hello"} for i in range(1000)]
    results = send_bulk_emails(email_list)
### upload_local_file(*file_paths)
Uploads files to Composio S3/R2 storage. Single files upload directly, multiple files are auto-zipped.
Use this when you need to upload/download any generated artifacts from the sandbox.
    def upload_local_file(*file_paths) -> tuple[Dict[str, Any], str]
    # Returns: (result_dict, error_string)
    # Success: ({"s3_url": str, "uploaded_file": str, "type": str, "id": str, "key": str, "message": str}, "")
    # Error: ({}, "error_message")
    # Single file
    result, error = upload_local_file("/path/to/report.pdf")
    # Multiple files (auto-zipped)
    result, error = upload_local_file("/home/user/doc1.txt", "/home/user/doc2.txt")
    # Always check for errors
    if error:
        print("Upload failed:", error)
        return
    print("Uploaded:", result["s3_url"])
    NOTE: Please do not forget to read the environment variables for input and end the notebook with just output (DO NOT PRINT)
  
    

Download a file from a public s3 (or r2) url to a local path.

Enable a specific trigger for the authenticated user.

Execute a tool using the composio api.

Executes a Vibe Api

Get the dependency graph for a given tool, showing related parent tools that might be useful. this action calls the composio labs dependency graph api to retrieve tools that are commonly used together with or before the specified tool. this helps discover related tools and understand common workflows.

Gets the required parameters for connecting to a toolkit via initiate connection. returns the exact parameter names and types needed for initiate connection's parameters field. supports api keys, oauth credentials, connection fields, and hybrid authentication scenarios. if has default credentials is true, you can call initiate connection with empty parameters for seamless oauth flow.

Retrieves the response schema for a specified composio tool. this action fetches the complete response schema definition for any valid composio tool, returning it as a dictionary that describes the expected response structure.

    Get the details of the existing vibe api for a given vibe api id.
    

Initiate a connection to a toolkit with comprehensive authentication support. supports all authentication scenarios: 1. composio default oauth (no parameters needed) 2. custom oauth (user's client id/client secret) 3. api key/bearer token authentication 4. basic auth (username/password) 5. hybrid scenarios (oauth + connection fields like site name) 6. connection-only fields (subdomain, api key at connection level) 7. no authentication required automatically detects and validates auth config vs connection fields, provides helpful error messages for missing parameters.

List all the available toolkits on composio with filtering options.

List available triggers and their configuration schemas.

  MCP Server Info: Composio MCP connects 500+ apps—Slack, GitHub, Notion, Google Workspace (Gmail, Sheets, Drive, Calendar), Microsoft (Outlook, Teams), X, Figma, Web Search, Meta apps (WhatsApp, Instagram), TikTok, AI tools like Nano Banana & Veo3, and more—for seamless cross-app automation. Use this MCP server to discover new tools and connect to apps.
<br />
Tool Info: Create/manage connections to user's apps. If COMPOSIO_SEARCH_TOOLS finds no active connection for an app, call this with the toolkit name and required auth params to create one. When the needed params are unclear, the tool returns a list—ask the user for each (e.g. API_KEY). ALWAYS show the list along with their descriptions(if possible) to the user.
If the response includes an OAuth redirect_url, ALWAYS show a FORMATTED MARKDOWN LINK to the user.
Supports OAuth (default/custom), API Key, Bearer Token, Basic Auth, hybrid, and no-auth. Batch-safe, isolates errors, allows selective re-init, returns per-app results and summary.
SESSION: ALWAYS pass session_id from previous meta tool calls. The session_id was in the COMPOSIO_SEARCH_TOOLS response. Required for workflow correlation.
    

  Fast and parallel tool executor for tools discovered through COMPOSIO_SEARCH_TOOLS. Use this tool to execute upto 20 tools in parallel across apps. Response contains structured outputs ready for immediate analysis - avoid reprocessing via remote bash/code execute tools.
  Prerequisites:
- Alwyas use valid tool slugs and their parameters discovered through COMPOSIO_SEARCH_TOOLS. NEVER invent tool slugs. ALWAYS pass arguments with the tool_slug in each tool.
- Active connection statuses for the tools that are going to be executed through COMPOSIO_MANAGE_CONNECTIONS.
- Ensure proper plan creation using COMPOSIO_CREATE_PLAN has been done.
- Cannot have any dependency of the response among the tools.
Usage guidelines:
- To be used whenever a tool is discovered and has to be called, either as part of a multi-step workflow or as a standalone tool.
- If COMPOSIO_SEARCH_TOOLS returns a tool that can perform the task, prefer calling it via this executor AFTER consulting COMPOSIO_CREATE_PLAN. Do not write custom API calls or ad-hoc scripts for tasks that can be completed by available Composio tools.
- Tools should be used highly parallelly.
- Predictively set sync_response_to_workbench=true if the response may be large or needed for later scripting. It still shows inline. However, if the actual response data turns out small and manageable without scripting, SKIP the workbench and use inline response directly.
- 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 along with summary.
- 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.
SESSION: ALWAYS pass session_id from previous meta tool calls. The session_id was in the COMPOSIO_SEARCH_TOOLS response. Required for workflow correlation.
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"]}
- INCORRECT format: {"slack": {"channel_id": "C1234567"}, "gmail": {"user": "john"}}
- 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, configuration details, user/resource identifiers that don't change frequently.
- DO NOT STORE: Action descriptions, temporary status updates, activity logs, "sent message" confirmations, "fetched data" descriptions.
- Examples of GOOD memory (store these):
  * "The important channel in Slack has ID C1234567 and is called #general"
  * "Venky's GitHub project 'MyProject' maps to project ID proj_abc123"
  * "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"
  * "Found 5 unread messages in inbox"
- Do not repeat the memories stored or found previously.
SESSION: ALWAYS pass session_id from previous meta tool calls. The session_id was in the COMPOSIO_SEARCH_TOOLS response. Required for workflow correlation.
    

  Execute bash commands in a REMOTE sandbox for file operations, data processing, and system tasks. Essential for handling large tool responses saved to remote files.
  PRIMARY USE CASES:
- Process large tool responses saved to remote sandbox
- File system operations, data processing with shell tools like jq, awk, sed, grep, etc.
WORKFLOW INTEGRATION:
1. IF COMPOSIO_MULTI_EXECUTE_TOOL saves large responses to remote sandbox
3. Extract specific information from JSON files using jq
IMPORTANT NOTES:
- Commands run from /home/user directory by default
SESSION: ALWAYS pass session_id from previous meta tool calls. The session_id was in the COMPOSIO_SEARCH_TOOLS response. Required for workflow correlation.
    

  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 Composio tool executions or proxy_execute calls (when no direct Composio tool exists for the task).
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 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 and, if too long, path(s) to artifacts under `/home/user/.code_out`.
IMPORTANT CODING RULES:
  1. Stepwise Execution: Prefer splitting work into small steps. Save intermediate outputs in variables or temporary file in `/tmp/`. Call COMPOSIO_REMOTE_WORKBENCH again for the next step. This improves composabilit and avoid timeouts.
  2. Notebook Persistence: This is a persistent Jupyter notebook cell: variables, functions, imports, and in-memory state from previous and future code executions are preserved in the notebook’s history and available for ruse. You also have a few helper functions available.
  3. Parallelism & Timeout (CRITICAL): There is a hard timeout of 4 minutes so complete the code within that. Prioritize PARALLEL execution using ThreadPoolExecutor with suitable concurrency for bulk operations - e.g., call run_composio_tool or invoke_llm parallelly across rows to maximize efficiency.
    3.1 If the data is large, split into smaller batches and call the workbench multiple times.
  4. Checkpoints: Implement checkpoints (in memory or files) so that long runs can be resumed from the last completed step.
  5. 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 via COMPOSIO_MULTI_EXECUTE_TOOL or use invoke_llm helper.
  6. LLM Helpers: You should always use invoke_llm helper for summary, analysis, or field extraction on results. This is a smart LLM that will give much better results than any adhoc filtering.
  7. Avoid Meta Loops: Do not use run_composio_tool to call COMPOSIO_MULTI_EXECUTE_TOOL or other COMPOSIO_* meta tools to avoid cycles. Only use it for app tools.
  8. Pagination: Use when data spans multiple pages. Continue fetching pages with the returned next_page_token or cursor until none remains. Parallelize fetching pages if tool supports page_number.
  9. No Hardcoding: Never hardcode data in code. Always load it from files or tool responses, iterating to construct intermediate or final inputs/outputs.
  10. Code Correctness (CRITICAL): Code must be syntactically and semantically correct and executable.
  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.
ENV & HELPERS:
1. Home directory: `/home/user`.
4. Helper functions initialized in the workbench:
    1) `run_composio_tool(tool_slug: str, arguments: dict)`: Execute a known Composio **app** tool (from COMPOSIO_SEARCH_TOOLS). Do not invent names; match the tool’s input schema. Suited for loops/parallel/bulk over datasets.
      1.1 run_composio_tools returns JSON with top-level "data". Parse carefully—structure may be nested.
    2) `invoke_llm(query: str)`: Invoke an LLM for semantic tasks. Pass MAX 400000 characters in input.
    3) `proxy_execute(method, endpoint, toolkit, params=None, body=None)`: Call a toolkit API directly when no Composio tool exists.
    4) `web_search(query: str)`: Search the web for information.
    5) `upload_local_file(*file_paths)`: Upload files to Composio S3/R2 storage. Use this to download any generated files/artifacts from the sandbox.
    6) `smart_file_extract(file_path: str)`: Extract text from any file format (pdf, image, doc, etc). Use this to extract content from workbench files in any format for summary/analysis. Use invoke_llm on response if it's large. Returns Tuple[str, str] (response, error)
    7) Workbench comes with comprehensive Image Processing (PIL/Pillow, OpenCV, scikit-image), PyTorch ML libraries, Document and Report handling tools (pandoc, python-docx, pdfplumber, reportlab), and standard Data Analysis tools (pandas, numpy, matplotlib) for advanced visual, analytical, and AI tasks."
## Python Helper Functions for LLM Scripting
### 1) run_composio_tool(tool_slug, arguments)
Executes a known Composio tool via backend API. Do NOT call COMPOSIO_* meta tools to avoid cyclic calls.
    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); return
    email_data = result.get("data", {})
    print("Fetched:", email_data)
### 2) invoke_llm(query)
Calls Groq LLM for reasoning, analysis, and semantic tasks. Pass MAX 400000 characters input.
    def invoke_llm(query: str) -> tuple[str, str]
    # Returns: (llm_response, error_message)
    resp, error = invoke_llm("Summarize the key points from this data")
    if error:
        print("invoke_llm error:", error); return
    print("LLM:", resp)
    # Example: analyze tool response with LLM
    tool_resp, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5, "user_id": "me"})
    if err:
        print("GMAIL_FETCH_EMAILS error:", err); return
    parsed = tool_resp.get("data", {})
    resp, err = invoke_llm(f"Analyze these emails and summarize: {parsed}")
    if err:
        print("invoke_llm error:", err); return
    print("LLM Gmail Summary:", resp)
    # TIP: batch prompts to reduce LLM calls.
### 4) proxy_execute(method, endpoint, toolkit, params=None, body=None)
Direct API call to a connected toolkit service.
    from typing import Literal, Optional
    def proxy_execute(
        method: Literal["GET","POST","PUT","DELETE","PATCH"],
        endpoint: str,
        toolkit: str,
        params: Optional[list[Parameter]] = None,
        body: Optional[object] = None,
    ) -> tuple[ToolProxyResponse | None, str]
    # Returns: (response_object, error_message)
    response, error = proxy_execute("GET", "https://api.github.com/repos/owner/repo", "github")
    if error:
        print("proxy_execute error:", error); return
    print("Success:", response.data)
### 5) web_search(query)
Searches the web via Exa AI.
    def web_search(query: str) -> tuple[str, str]
    # Returns: (search_results_text, error_message)
    results, error = web_search("latest developments in AI")
    if error:
        print("web_search error:", error)
    else:
        print("Results:", results)
## Best Practices
### Error-first pattern
    data, error = some_helper(...)
    if error:
        print("some_helper error:", error); return
    # safe to use `data`
### Defensive parsing (print keys while narrowing)
    res, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5})
    if not err and 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 (4-min sandbox timeout)
Adjust concurrency so all tasks finish within 4 minutes.
    import concurrent.futures
    MAX_CONCURRENCY = 10 # Adjust as needed
    def send_bulk_emails(email_list):
        def send_single(email):
            result, error = run_composio_tool("GMAIL_SEND_EMAIL", {
                "to": email["recipient"], "subject": email["subject"], "body": email["body"]
            })
            if error:
                print(f"Failed {email['recipient']}: {error}")
                return {"status": "failed", "error": error}
            return {"status": "sent", "data": result}
        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as ex:
            futures = [ex.submit(send_single, e) for e in email_list]
            for f in concurrent.futures.as_completed(futures):
                results.append(f.result())
        return results
    # Example usage
    email_list = [{"recipient": f"user{i}@example.com", "subject": "Test", "body": "Hello"} for i in range(1000)]
    results = send_bulk_emails(email_list)
### upload_local_file(*file_paths)
Uploads files to Composio S3/R2 storage. Single files upload directly, multiple files are auto-zipped.
Use this when you need to upload/download any generated artifacts from the sandbox.
    def upload_local_file(*file_paths) -> tuple[Dict[str, Any], str]
    # Returns: (result_dict, error_string)
    # Success: ({"s3_url": str, "uploaded_file": str, "type": str, "id": str, "key": str, "message": str}, "")
    # Error: ({}, "error_message")
    # Single file
    result, error = upload_local_file("/path/to/report.pdf")
    # Multiple files (auto-zipped)
    result, error = upload_local_file("/home/user/doc1.txt", "/home/user/doc2.txt")
    # Always check for errors
    if error:
        print("Upload failed:", error)
        return
    print("Uploaded:", result["s3_url"])
Guidance: Ensure to peform the task with High Accuracy and Completeness. For large data, use parallel processing (ThreadPoolExecutor) and fewer batches in each call to maximise efficiency. Leverage invoke_llm for smart analysis whenever needed. NEVER hardcode data in code and NEVER run COMPOSIO_MULTI_EXECUTE_TOOL in the workbench.
Memory Storage:
- 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.
- 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"]}
- INCORRECT format: {"slack": {"channel_id": "C1234567"}, "gmail": {"user": "john"}}
- 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, configuration details, user/resource identifiers that don't change frequently.
- DO NOT STORE: Action descriptions, temporary status updates, activity logs, "sent message" confirmations, "fetched data" descriptions.
- Examples of GOOD memory (store these):
  * "The important channel in Slack has ID C1234567 and is called #general"
  * "Venky's GitHub project 'MyProject' maps to project ID proj_abc123"
  * "The team's main repository is owned by user 'teamlead' with ID 98765"
- 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"
  * "Found 5 unread messages in inbox"
    

  MCP Server Info: Composio MCP connects 500+ apps—Slack, GitHub, Notion, Google Workspace (Gmail, Sheets, Drive, Calendar), Microsoft (Outlook, Teams), X, Figma, Web Search, Meta apps (WhatsApp, Instagram), TikTok, AI tools like Nano Banana & Veo3, and more—for seamless cross-app automation.
  Use this MCP server to discover new tools and connect to apps.
  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.
<br />
  Tool Info: Extremely fast search tool to discover available MCP callable tools that can be used to solve a particular problem, user query or complete a task.
Usage guidelines:
  <recommended>
  - Use this tool whenever kicking off a task. Post this, keep coming back to this tool to discover new tools.
  - If the user pivots to a different use case in same chat, you MUST call this tool again with the new use case.
  </recommended>
  - Specify the use_case with a detailed description of the problem, query, or task. Be clear and precise so the system can find the most relevant tools. Queries can involve one or multiple apps, and be simple or complex—from a quick action to a multi-step, cross-app workflow.
  - Pass known_fields as a list of key-value pairs to help the search provide tools to look up missing details (for example, finding channel_id from a given channel_name).
  - To understand the abilities and skills at hand, set exploratory_query to true. Otherwise, keep it as false for task specific searches.
  - After this tool, call COMPOSIO_CREATE_PLAN to ensure proper plan creation.
Response:
  - The response is a list of toolkits (apps) and tools suitable for the task, along with their tool_slug, description, input schema, and related tools that may serve as prerequisites, fallbacks, or next steps. It also includes a recommended order of execution and a brief reasoning for why each result was returned.
  - If a toolkit has an active connection, the response includes it along with any available current user information. If no active connection exists, the response lists any required parameters for establishing a new one.
  - The response includes the current UTC time for reference. You can reference UTC time from the response if needed.
  - 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 going forward.
  - The tools returned to you through this are to be called via COMPOSIO_MULTI_EXECUTE_TOOL. Make sure to specify the tool_slug and arguments for each tool execution properly.
SESSION: ALWAYS call 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.
    

    Update the existing vibe api with the new code and input/output schemas.
    
    This tool allows you to:
    1. Save the full workflow execution as a reusable notebook
    2. If I share the notebook with others (when published) it can be used by passing correct environment variables and it executes the whole workflow from beginning of this session
    3. You should generate input json schema of this notebook based on the executed workflow, so that other users of this published notebook can pass their own valid inputs
    4. You should generate the code of this notebook based on the executed workflow, please see below for more instructions on how to generate the code for the notebook
    5. Similarly please generate good output json schema for this notebook, so that other users of this published notebook know how to consume the response
    6. The notebook should take the input from environment variables using os.environ.get(). The user of this notebook will pass each key of the input json schema as an env var
    WHEN TO USE
    - Only run this tool when the workflow is completed and successful or if the user explicitly asked to run this tool
    DO NOT USE
    - When the workflow is still being processed, or not yet completed and the user explicitly didn't ask to run this tool
    IMPORTANT CODING RULES:
    1. Single Execution: Please generate the code for the full notebook that can be executed in a single invocation
    2. 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 via COMPOSIO_MULTI_EXECUTE_TOOL or use invoke_llm helper.
    3. Parallelism & Timeout (CRITICAL): There is a hard timeout of 4 minutes so complete the code within that. Prioritize PARALLEL execution using ThreadPoolExecutor with suitable concurrency for bulk operations - e.g., call run_composio_tool or invoke_llm parallelly across rows to maximize efficiency.
    4. LLM Helpers: You should always use invoke_llm helper for summary, analysis, or field extraction on results. This is a smart LLM that will give much better results than any adhoc filtering.
    5. Avoid Meta Loops: Do not use run_composio_tool to call COMPOSIO_MULTI_EXECUTE_TOOL or other COMPOSIO_* meta tools to avoid cycles. Only use it for app tools.
    6. Pagination: Use when data spans multiple pages. Continue fetching pages with the returned next_page_token or cursor until none remains. Parallelize fetching pages if tool supports page_number.
    7. No Hardcoding: Never hardcode data in code. Always load it from files or tool responses, iterating to construct intermediate or final inputs/outputs.
    8. Code Correctness (CRITICAL): Code must be syntactically and semantically correct and executable.
    9. Please ensure that your code takes input in the form of input json schema via environment variables and stores the output at last command in the form of output json schema
    10. Your notebook must read each key of the input json schema in your code from environment variables using os.environ.get()
    11. Please always end the code with just "output" command following the output json schema so that the notebook actually shows it. DO NOT PRINT IT AT ANY COST
    12. Please only take needed args as input in input json schema and only give the needed data in the output. This is to keep the json schemas and notebook simple
    13. Debugging (CRITICAL): For every print statement in your code, please prefix it with the time at which it is being executed. This will help me investigate latency related stuff
    ENV & HELPERS:
    1. Home directory: `/home/user`.
    2. Helper functions initialized in the sandbox where the notebook is going to run:
      1) `run_composio_tool(tool_slug: str, arguments: dict)`: Execute a known Composio **app** tool (from COMPOSIO_SEARCH_TOOLS). Do not invent names; match the tool’s input schema. Suited for loops/parallel/bulk over datasets.
        1.1) run_composio_tools returns JSON with top-level "data". Parse carefully—structure may be nested.
      2) `invoke_llm(query: str)`: Invoke an LLM for semantic tasks. Pass MAX 400000 characters in input.
      3) `proxy_execute(method, endpoint, toolkit, params=None, body=None)`: Call a toolkit API directly when no Composio tool exists.
      4) `web_search(query: str)`: Search the web for information.
      5) `upload_local_file(*file_paths)`: Upload files to Composio S3/R2 storage. Use this to download any generated files/artifacts from the sandbox.
      6) `smart_file_extract(file_path: str)`: Extract text from any file format (pdf, image, doc, etc). Use this to extract content from workbench files in any format for summary/analysis. Use invoke_llm on response if it's large. Returns Tuple[str, str] (response, error)
    
## Python Helper Functions for LLM Scripting
### 1) run_composio_tool(tool_slug, arguments)
Executes a known Composio tool via backend API. Do NOT call COMPOSIO_* meta tools to avoid cyclic calls.
    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); return
    email_data = result.get("data", {})
    print("Fetched:", email_data)
### 2) invoke_llm(query)
Calls Groq LLM for reasoning, analysis, and semantic tasks. Pass MAX 400000 characters input.
    def invoke_llm(query: str) -> tuple[str, str]
    # Returns: (llm_response, error_message)
    resp, error = invoke_llm("Summarize the key points from this data")
    if error:
        print("invoke_llm error:", error); return
    print("LLM:", resp)
    # Example: analyze tool response with LLM
    tool_resp, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5, "user_id": "me"})
    if err:
        print("GMAIL_FETCH_EMAILS error:", err); return
    parsed = tool_resp.get("data", {})
    resp, err = invoke_llm(f"Analyze these emails and summarize: {parsed}")
    if err:
        print("invoke_llm error:", err); return
    print("LLM Gmail Summary:", resp)
    # TIP: batch prompts to reduce LLM calls.
### 4) proxy_execute(method, endpoint, toolkit, params=None, body=None)
Direct API call to a connected toolkit service.
    from typing import Literal, Optional
    def proxy_execute(
        method: Literal["GET","POST","PUT","DELETE","PATCH"],
        endpoint: str,
        toolkit: str,
        params: Optional[list[Parameter]] = None,
        body: Optional[object] = None,
    ) -> tuple[ToolProxyResponse | None, str]
    # Returns: (response_object, error_message)
    response, error = proxy_execute("GET", "https://api.github.com/repos/owner/repo", "github")
    if error:
        print("proxy_execute error:", error); return
    print("Success:", response.data)
### 5) web_search(query)
Searches the web via Exa AI.
    def web_search(query: str) -> tuple[str, str]
    # Returns: (search_results_text, error_message)
    results, error = web_search("latest developments in AI")
    if error:
        print("web_search error:", error)
    else:
        print("Results:", results)
## Best Practices
### Error-first pattern
    data, error = some_helper(...)
    if error:
        print("some_helper error:", error); return
    # safe to use `data`
### Defensive parsing (print keys while narrowing)
    res, err = run_composio_tool("GMAIL_FETCH_EMAILS", {"max_results": 5})
    if not err and 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 (4-min sandbox timeout)
Adjust concurrency so all tasks finish within 4 minutes.
    import concurrent.futures
    MAX_CONCURRENCY = 10 # Adjust as needed
    def send_bulk_emails(email_list):
        def send_single(email):
            result, error = run_composio_tool("GMAIL_SEND_EMAIL", {
                "to": email["recipient"], "subject": email["subject"], "body": email["body"]
            })
            if error:
                print(f"Failed {email['recipient']}: {error}")
                return {"status": "failed", "error": error}
            return {"status": "sent", "data": result}
        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as ex:
            futures = [ex.submit(send_single, e) for e in email_list]
            for f in concurrent.futures.as_completed(futures):
                results.append(f.result())
        return results
    # Example usage
    email_list = [{"recipient": f"user{i}@example.com", "subject": "Test", "body": "Hello"} for i in range(1000)]
    results = send_bulk_emails(email_list)
### upload_local_file(*file_paths)
Uploads files to Composio S3/R2 storage. Single files upload directly, multiple files are auto-zipped.
Use this when you need to upload/download any generated artifacts from the sandbox.
    def upload_local_file(*file_paths) -> tuple[Dict[str, Any], str]
    # Returns: (result_dict, error_string)
    # Success: ({"s3_url": str, "uploaded_file": str, "type": str, "id": str, "key": str, "message": str}, "")
    # Error: ({}, "error_message")
    # Single file
    result, error = upload_local_file("/path/to/report.pdf")
    # Multiple files (auto-zipped)
    result, error = upload_local_file("/home/user/doc1.txt", "/home/user/doc2.txt")
    # Always check for errors
    if error:
        print("Upload failed:", error)
        return
    print("Uploaded:", result["s3_url"])
    NOTE: Please do not forget to read the environment variables for input and end the notebook with just output (DO NOT PRINT)
  
    

Wait for the user to complete authentication AFTER you have given them an auth URL from COMPOSIO_MANAGE_CONNECTIONS. Use this **immediately after** sharing the auth URL so you can automatically continue once the connection is established — without waiting for the user to manually come back and say they’re done. Make sure you have printed the auth URL, DO NOT call this tool without printing auth URL first.
    This ensures a smooth, uninterrupted flow and a better user experience.
  You NEED NOT wait if there is no auth URL in the response of COMPOSIO_MANAGE_CONNECTIONS like in cases you ask user for  api_key, client_id or client_secret.
    Input params <toolkits: list of toolkit names>, <mode (any / all) : wait for ANY or ALL connections to reach success/failed state (default: any) >
    Output params <connection statuses>
    Example:
    input:
    toolkits: [gmail, outlook]
    mode: [any]
    output: {
      gmail: {
        status: [connected]
      },
      outlook: {
        status: [initiated]
      }
    }
SESSION: ALWAYS pass session_id from previous meta tool calls. The session_id was in the COMPOSIO_SEARCH_TOOLS response. Required for workflow correlation.