Surface output artifact_id for operation chaining (#246)

Thread the output artifact_id from the API through progress messages
and result responses so clients can chain operations (e.g.
upload → agent → rank → screen) without re-uploading data.

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: RafaelPo

everyrow-mcp/src/everyrow_mcp/result_store.py

@@ -95,6 +95,7 @@ def _build_result_response(
     session_url: str = "",
     poll_token: str = "",
     mcp_server_url: str = "",
+    artifact_id: str = "",
     *,
     requested_page_size: int | None = None,
     skip_widget: bool = False,
@@ -148,6 +149,8 @@ def _build_result_response(
         }
         if session_url:
             widget_data["session_url"] = session_url
+        if artifact_id:
+            widget_data["artifact_id"] = artifact_id
         if poll_token:
             widget_data["poll_token"] = poll_token
             widget_data["download_token_url"] = (
@@ -182,6 +185,9 @@ def _build_result_response(
             f"of {total} (final page)."
         )
 
+    if artifact_id:
+        summary += f"\nOutput artifact_id (use to chain into next tool): {artifact_id}"
+
     contents.append(TextContent(type="text", text=summary))
     return contents
 
@@ -265,6 +271,7 @@ async def try_cached_result(
         session_url=meta.get("session_url", ""),
         poll_token=poll_token or "",
         mcp_server_url=mcp_server_url,
+        artifact_id=meta.get("artifact_id", ""),
         requested_page_size=page_size,
         skip_widget=skip_widget,
         skip_session=skip_session,
@@ -278,6 +285,7 @@ async def try_store_result(
     page_size: int,
     session_url: str = "",
     mcp_server_url: str = "",
+    artifact_id: str = "",
     *,
     skip_widget: bool = False,
     skip_session: bool = False,
@@ -294,6 +302,8 @@ async def try_store_result(
         meta: dict[str, Any] = {"total": total, "columns": columns}
         if session_url:
             meta["session_url"] = session_url
+        if artifact_id:
+            meta["artifact_id"] = artifact_id
         await redis_store.store_result_meta(task_id, json.dumps(meta))
 
         # Build and cache page preview
@@ -328,6 +338,7 @@ async def try_store_result(
             session_url=session_url,
             poll_token=poll_token or "",
             mcp_server_url=mcp_server_url,
+            artifact_id=artifact_id,
             requested_page_size=page_size,
             skip_widget=skip_widget,
             skip_session=skip_session,

everyrow-mcp/src/everyrow_mcp/tool_helpers.py

@@ -373,6 +373,14 @@ def total(self) -> int:
         p = self._response.progress
         return p.total if p else 0
 
+    @computed_field
+    @property
+    def artifact_id(self) -> str:
+        aid = self._response.artifact_id
+        if aid is not None and not isinstance(aid, Unset):
+            return str(aid)
+        return ""
+
     @computed_field
     @property
     def error(self) -> str | None:
@@ -443,6 +451,8 @@ def progress_message(self, task_id: str) -> str:
                              After the user responds, call everyrow_results(task_id='{task_id}', page_size=N).""")
                 else:
                     next_call = f"Call everyrow_results(task_id='{task_id}', output_path='<choose_a_path>.csv') to save the output."
+                if self.artifact_id:
+                    completed_msg += f"\nOutput artifact_id: {self.artifact_id}"
                 return f"{completed_msg}\n{next_call}"
             return f"Task {self.status.value}. Report the error to the user."
 
@@ -535,13 +545,15 @@ def __init__(self, status: str) -> None:
         super().__init__(status)
 
 
-async def _fetch_task_result(client: Any, task_id: str) -> tuple[pd.DataFrame, str]:
-    """Fetch a task's result DataFrame and session ID from the API.
+async def _fetch_task_result(
+    client: Any, task_id: str
+) -> tuple[pd.DataFrame, str, str]:
+    """Fetch a task's result DataFrame, session ID, and output artifact ID from the API.
 
     Checks task status first, then retrieves and parses the result data.
 
     Returns:
-        Tuple of (DataFrame, session_id).
+        Tuple of (DataFrame, session_id, artifact_id).
 
     Raises:
         TaskNotReady: If the task is not in a terminal state.
@@ -576,9 +588,18 @@ async def _fetch_task_result(client: Any, task_id: str) -> tuple[pd.DataFrame, s
         )
     )
 
+    artifact_id = ""
+    aid = result_response.artifact_id
+    if aid is not None and not isinstance(aid, Unset):
+        artifact_id = str(aid)
+
     if isinstance(result_response.data, list):
         records = [item.additional_properties for item in result_response.data]
-        return pd.DataFrame(records), session_id
+        return pd.DataFrame(records), session_id, artifact_id
     if isinstance(result_response.data, TaskResultResponseDataType1):
-        return pd.DataFrame([result_response.data.additional_properties]), session_id
+        return (
+            pd.DataFrame([result_response.data.additional_properties]),
+            session_id,
+            artifact_id,
+        )
     raise ValueError("Task result has no table data.")

everyrow-mcp/src/everyrow_mcp/tools.py

@@ -844,7 +844,7 @@ async def everyrow_results_stdio(
     task_id = params.task_id
 
     try:
-        df, _session_id = await _fetch_task_result(client, task_id)
+        df, _session_id, artifact_id = await _fetch_task_result(client, task_id)
     except TaskNotReady as e:
         return [
             TextContent(
@@ -865,11 +865,12 @@ async def everyrow_results_stdio(
 
     output_file = Path(params.output_path)
     save_result_to_csv(df, output_file)
+    artifact_line = f"\nOutput artifact_id: {artifact_id}" if artifact_id else ""
     return [
         TextContent(
             type="text",
             text=dedent(f"""\
-                Saved {len(df)} rows to {output_file}
+                Saved {len(df)} rows to {output_file}{artifact_line}
 
                 Tip: For multi-step pipelines or custom response models, \
                 use the everyrow Python SDK directly."""),
@@ -924,7 +925,7 @@ async def everyrow_results_http(
 
     # ── Fetch from API ────────────────────────────────────────────
     try:
-        df, session_id = await _fetch_task_result(client, task_id)
+        df, session_id, artifact_id = await _fetch_task_result(client, task_id)
         session_url = get_session_url(UUID(session_id)) if session_id else ""
     except TaskNotReady as e:
         return [
@@ -955,6 +956,7 @@ async def everyrow_results_http(
         params.page_size,
         session_url,
         mcp_server_url=mcp_server_url,
+        artifact_id=artifact_id,
         skip_widget=skip_widget,
         skip_session=skip_session,
     )

everyrow-mcp/tests/test_result_store.py

@@ -251,6 +251,42 @@ def test_no_poll_token_when_empty(self):
         assert "poll_token" not in widget
         assert "download_token_url" not in widget
 
+    def test_artifact_id_included_in_widget_and_text(self):
+        """When artifact_id is provided, it appears in widget JSON and text summary."""
+        preview = [{"a": 1}]
+        csv_url = f"{FAKE_SERVER_URL}/api/results/task-aid/download?token=abc"
+        result = _build_result_response(
+            task_id="task-aid",
+            csv_url=csv_url,
+            preview_records=preview,
+            total=1,
+            columns=["a"],
+            offset=0,
+            page_size=10,
+            artifact_id="abc-123-def",
+        )
+        widget = json.loads(result[0].text)
+        assert widget["artifact_id"] == "abc-123-def"
+        assert "abc-123-def" in result[1].text
+        assert "Output artifact_id" in result[1].text
+
+    def test_no_artifact_id_when_empty(self):
+        """When artifact_id is empty, widget JSON and text omit it."""
+        preview = [{"a": 1}]
+        csv_url = f"{FAKE_SERVER_URL}/api/results/task-noaid/download?token=abc"
+        result = _build_result_response(
+            task_id="task-noaid",
+            csv_url=csv_url,
+            preview_records=preview,
+            total=1,
+            columns=["a"],
+            offset=0,
+            page_size=10,
+        )
+        widget = json.loads(result[0].text)
+        assert "artifact_id" not in widget
+        assert "Output artifact_id" not in result[1].text
+
 
 # ── Async functions ────────────────────────────────────────────
 
@@ -325,6 +361,29 @@ async def test_preserves_session_url_from_meta(self, _http_state):
         widget = json.loads(result[0].text)
         assert widget["session_url"] == "https://everyrow.io/sessions/xyz"
 
+    @pytest.mark.asyncio
+    async def test_preserves_artifact_id_from_meta(self, _http_state):
+        meta = json.dumps(
+            {
+                "total": 1,
+                "columns": ["a"],
+                "artifact_id": "cached-artifact-789",
+            }
+        )
+        page = json.dumps([{"a": 1}])
+        task_id = "task-cached-aid"
+
+        await redis_store.store_result_meta(task_id, meta)
+        await redis_store.store_result_page(task_id, 0, 10, page)
+        await redis_store.store_poll_token(task_id, "test-token")
+
+        result = await try_cached_result(task_id, 0, 10, mcp_server_url=FAKE_SERVER_URL)
+
+        assert result is not None
+        widget = json.loads(result[0].text)
+        assert widget["artifact_id"] == "cached-artifact-789"
+        assert "cached-artifact-789" in result[1].text
+
     @pytest.mark.asyncio
     async def test_returns_none_when_json_expired(self, _http_state):
         """When metadata exists but JSON is gone, fall back to API (return None)."""
@@ -383,6 +442,35 @@ async def test_stores_and_returns_response(self, sample_df, _http_state):
         assert meta["total"] == 3
         assert meta["columns"] == ["name", "score"]
 
+    @pytest.mark.asyncio
+    async def test_stores_artifact_id_in_meta_and_response(
+        self, sample_df, _http_state
+    ):
+        task_id = "task-aid-store"
+        await redis_store.store_poll_token(task_id, "test-token")
+
+        result = await try_store_result(
+            task_id,
+            sample_df,
+            0,
+            10,
+            mcp_server_url=FAKE_SERVER_URL,
+            artifact_id="output-artifact-456",
+        )
+
+        # Verify artifact_id in Redis metadata
+        meta_raw = await redis_store.get_result_meta(task_id)
+        assert meta_raw is not None
+        meta = json.loads(meta_raw)
+        assert meta["artifact_id"] == "output-artifact-456"
+
+        # Verify artifact_id in widget JSON
+        widget = json.loads(result[0].text)
+        assert widget["artifact_id"] == "output-artifact-456"
+
+        # Verify artifact_id in text summary
+        assert "output-artifact-456" in result[1].text
+
     @pytest.mark.asyncio
     async def test_includes_session_url_in_meta(self, sample_df, _http_state):
         task_id = "task-sess"

everyrow-mcp/tests/test_tool_helpers.py

@@ -0,0 +1,117 @@
+"""Tests for tool_helpers.py — TaskState, _fetch_task_result, progress_message."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+from uuid import uuid4
+
+from everyrow.generated.models.public_task_type import PublicTaskType
+from everyrow.generated.models.task_status import TaskStatus
+from everyrow.generated.types import UNSET
+
+from everyrow_mcp.tool_helpers import TaskState
+from tests.conftest import override_settings
+
+
+def _make_status_response(
+    *,
+    status: TaskStatus = TaskStatus.COMPLETED,
+    task_type: PublicTaskType = PublicTaskType.AGENT,
+    artifact_id=UNSET,
+    session_id=None,
+    error=UNSET,
+    progress=None,
+    created_at=None,
+    updated_at=None,
+):
+    """Build a mock TaskStatusResponse."""
+    resp = MagicMock()
+    resp.status = status
+    resp.task_type = task_type
+    resp.artifact_id = artifact_id
+    resp.session_id = session_id
+    resp.error = error
+    resp.created_at = created_at
+    resp.updated_at = updated_at
+
+    if progress is None:
+        p = MagicMock()
+        p.completed = 5
+        p.failed = 0
+        p.running = 0
+        p.total = 5
+        resp.progress = p
+    else:
+        resp.progress = progress
+
+    return resp
+
+
+class TestTaskStateArtifactId:
+    def test_artifact_id_from_uuid(self):
+        uid = uuid4()
+        resp = _make_status_response(artifact_id=uid)
+        ts = TaskState(resp)
+        assert ts.artifact_id == str(uid)
+
+    def test_artifact_id_unset(self):
+        resp = _make_status_response(artifact_id=UNSET)
+        ts = TaskState(resp)
+        assert ts.artifact_id == ""
+
+    def test_artifact_id_none(self):
+        resp = _make_status_response(artifact_id=None)
+        ts = TaskState(resp)
+        assert ts.artifact_id == ""
+
+
+class TestProgressMessageArtifactId:
+    def test_completed_message_includes_artifact_id(self):
+        uid = uuid4()
+        resp = _make_status_response(
+            status=TaskStatus.COMPLETED,
+            artifact_id=uid,
+        )
+        ts = TaskState(resp)
+        msg = ts.progress_message("task-123")
+        assert f"Output artifact_id: {uid}" in msg
+
+    def test_completed_message_omits_artifact_id_when_absent(self):
+        resp = _make_status_response(
+            status=TaskStatus.COMPLETED,
+            artifact_id=UNSET,
+        )
+        ts = TaskState(resp)
+        msg = ts.progress_message("task-123")
+        assert "Output artifact_id" not in msg
+
+    def test_completed_http_mode_includes_artifact_id(self):
+        uid = uuid4()
+        resp = _make_status_response(
+            status=TaskStatus.COMPLETED,
+            artifact_id=uid,
+        )
+        ts = TaskState(resp)
+        with override_settings(transport="streamable-http"):
+            msg = ts.progress_message("task-456")
+        assert f"Output artifact_id: {uid}" in msg
+        assert "everyrow_results" in msg
+
+    def test_running_message_does_not_include_artifact_id(self):
+        uid = uuid4()
+        resp = _make_status_response(
+            status=TaskStatus.RUNNING,
+            artifact_id=uid,
+        )
+        ts = TaskState(resp)
+        msg = ts.progress_message("task-789")
+        assert "Output artifact_id" not in msg
+
+    def test_failed_message_does_not_include_artifact_id(self):
+        resp = _make_status_response(
+            status=TaskStatus.FAILED,
+            error="Something went wrong",
+        )
+        ts = TaskState(resp)
+        msg = ts.progress_message("task-err")
+        assert "Output artifact_id" not in msg