diff --git a/actions/agent-connector/CHANGELOG.md b/actions/agent-connector/CHANGELOG.md index 9e0283b2..a5fa88b3 100644 --- a/actions/agent-connector/CHANGELOG.md +++ b/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 diff --git a/actions/agent-connector/README.md b/actions/agent-connector/README.md index a72cd6f7..42898488 100644 --- a/actions/agent-connector/README.md +++ b/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: diff --git a/actions/agent-connector/actions.py b/actions/agent-connector/actions.py index 298cda48..1c8440a8 100644 --- a/actions/agent-connector/actions.py +++ b/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) diff --git a/actions/agent-connector/package.yaml b/actions/agent-connector/package.yaml index 88cf10c5..eb41ddde 100644 --- a/actions/agent-connector/package.yaml +++ b/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