Agent Connector 4.3.3: Add create_work_items_from_dataframe action

actions/agent-connector/CHANGELOG.md

@@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](https://semver.org/).
 
+### [4.3.3] - 2026-05-08
+
+### Added
+
+- New `create_work_items_from_dataframe()` action: creates one Work Item per row of a named dataframe for a specified agent
+
+### Fixed
+
+- `create_work_items_from_dataframe()` works across all server versions by bypassing the library's `get_data_frame()`, which is incompatible with servers that return dataframe rows as a list of dicts instead of `{"columns": [...], "rows": [[...]]}`:
+  - Tries `threads/{thread_id}/data-frames/{name}` first (newer local servers)
+  - Falls back to `data-frames/{name}` (cloud servers, where thread context is carried via the invocation header)
+  - Logs the full URL attempted and outcome for each candidate path
+
 ### [4.3.1] - 2026-05-05
 
 ### Fixed

actions/agent-connector/README.md

@@ -30,6 +30,7 @@ The easiest way to interact with agents is using the `ask_agent` function:
 **Work Items:**
 - Create a Work Item for an agent by name
 - Create multiple Work Items in a single batch call
+- Create one Work Item per row from a dataframe available in the current thread
 - Get the current conversation ID (for passing back to a parent agent)
 
 ## Example Usage
@@ -96,6 +97,18 @@ Create work items for "Invoice Agent":
 
 The `message` field is automatically merged into the payload so the worker receives it as `payload["message"]`.
 
+### Create Work Items From a Dataframe
+
+Use `create_work_items_from_dataframe` to dispatch one Work Item per row of a dataframe that is available in the current thread. Each row becomes the payload for a separate Work Item.
+
+```
+Create work items for "Invoice Agent" using the dataframe "invoices"
+```
+
+> Created 5 work items for "Invoice Agent" from dataframe "invoices".
+
+The column names become the payload keys, so a dataframe with columns `invoice_id`, `amount`, `due_date` will produce payloads like `{"invoice_id": "IN-001", "amount": 250.00, "due_date": "2026-06-01"}`.
+
 ### Pass the Current Conversation ID to Worker Agents
 
 When orchestrating workers, use `get_current_conversation_id` to retrieve the calling conversation's ID and include it in the work item payload so workers can send replies back:

actions/agent-connector/actions.py

@@ -452,3 +452,112 @@ def create_work_item_for_agent(
             agent_id=agent.id,
         )
     )
+
+
+def _fetch_dataframe(name: str):
+    """Fetch a dataframe from the agent server, handling multiple server versions.
+
+    Different server versions use different endpoint paths and response formats:
+    - Newer servers: threads/{thread_id}/data-frames/{name} → list of row dicts
+    - Cloud servers: data-frames/{name} → list of row dicts (thread from context header)
+    - Older servers: data-frames/{name} → {"columns": [...], "rows": [[...]]}
+    """
+    from sema4ai.actions import Table
+    from sema4ai.actions.agent._client import (
+        _AgentAPIClient as _AgentServerClient,
+        AgentApiClientException,
+    )
+
+    thread_id = get_thread_id()
+    server_client = _AgentServerClient()
+
+    # Try thread-scoped path first (newer local servers), then fall back to
+    # the flat path (cloud/older servers that use invocation context for thread).
+    candidates = [
+        f"threads/{thread_id}/data-frames/{name}",
+        f"data-frames/{name}",
+    ]
+    last_exc = None
+    data = None
+    for path in candidates:
+        full_url = f"{server_client.api_url}{path}"
+        print(f"[_fetch_dataframe] GET {full_url}")
+        try:
+            response = server_client.request(path, method="GET")
+            data = response.json()
+            print(f"[_fetch_dataframe] success: {path}, response type={type(data).__name__}, len={len(data) if isinstance(data, (list, dict)) else 'n/a'}")
+            break
+        except AgentApiClientException as e:
+            print(f"[_fetch_dataframe] failed: {path} → {e}")
+            if e.status_code != 404:
+                raise ActionError(f"Failed to fetch dataframe '{name}': {e}") from e
+            last_exc = e
+
+    if data is None:
+        raise ActionError(f"Dataframe '{name}' not found") from last_exc
+
+    if isinstance(data, dict):
+        return Table(
+            columns=data["columns"],
+            rows=data["rows"],
+            name=data.get("name"),
+            description=data.get("description"),
+        )
+    if isinstance(data, list):
+        if not data:
+            return Table(columns=[], rows=[])
+        first = data[0]
+        if isinstance(first, dict):
+            columns = list(first.keys())
+            rows = [[row.get(col) for col in columns] for row in data]
+        else:
+            rows = data
+            columns = [str(i) for i in range(len(first) if first else 0)]
+        return Table(columns=columns, rows=rows)
+    raise ActionError(f"Unexpected response format for dataframe '{name}'")
+
+
+@action
+def create_work_items_from_dataframe(
+    dataframe_name: str,
+    agent_name: str,
+    sema4_api_key: Secret,
+    sema4_api_url: Secret,
+) -> Response[list[WorkItemResponse]]:
+    """Creates one Work Item per row of a named dataframe for a specific agent.
+
+    Args:
+        dataframe_name: Name of the dataframe available in the current thread
+        agent_name: Name of the agent to receive the work items
+        sema4_api_key: The API key for the Sema4 API. Use LOCAL if in Studio or SDK!
+        sema4_api_url: The base URL for the Sema4 API. Use LOCAL if in Studio or SDK!
+
+    Returns:
+        Response containing a list of created Work Item details, one per row
+    """
+    table = _fetch_dataframe(dataframe_name)
+
+    client = _make_client(sema4_api_key, sema4_api_url)
+    agent_result = resolve_agent_by_name(client, agent_name)
+    if not agent_result.found:
+        raise ActionError(agent_result.message)
+
+    wi_url = sema4_api_url.value if sema4_api_url.value.upper() != "LOCAL" else None
+    agent = agent_result.agent
+
+    results = []
+    for row in table.rows:
+        payload = dict(zip(table.columns, row))
+        work_item = client.create_work_item(
+            agent_id=agent.id,
+            payload=payload,
+            work_item_api_url=wi_url,
+        )
+        results.append(
+            WorkItemResponse(
+                work_item=work_item,
+                agent_name=agent.name,
+                agent_id=agent.id,
+            )
+        )
+    return Response(result=results)

actions/agent-connector/package.yaml

@@ -5,7 +5,7 @@ name: Agent Connector
 description: Actions to connect agents with each other
 
 # Package version number, recommend using semver.org
-version: 4.3.1
+version: 4.3.3
 
 # The version of the `package.yaml` format.
 spec-version: v2