From 75ab9d850520fa3e9512f77230846f7d75bff742 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Fri, 3 Jul 2026 00:14:07 +0200 Subject: [PATCH 1/8] fix: rename env var from PIPELEX_API_URL to PIPELEX_BASE_URL for consistency chore: update pipelex-sdk to version 0.2.0 and adjust related documentation --- .env.example | 4 ++-- CHANGELOG.md | 4 ++++ CLAUDE.md | 2 +- README.md | 10 +++++----- my_project/hello_world.py | 8 ++++---- pyproject.toml | 2 +- tests/conftest.py | 2 +- tests/integration/test_fundamentals.py | 4 ++-- uv.lock | 14 +++++++------- 9 files changed, 27 insertions(+), 23 deletions(-) diff --git a/.env.example b/.env.example index 8f75d49..d5e467c 100644 --- a/.env.example +++ b/.env.example @@ -1,8 +1,8 @@ # Pipelex API endpoint and credentials. # Sign up for an API key at https://go.pipelex.com/waitlist -PIPELEX_API_URL=https://api.pipelex.com +PIPELEX_BASE_URL=https://api.pipelex.com PIPELEX_API_KEY= # Self-hosted (open-source) runner instead of the hosted API: -# PIPELEX_API_URL=http://127.0.0.1:8081 +# PIPELEX_BASE_URL=http://127.0.0.1:8081 # PIPELEX_API_KEY= diff --git a/CHANGELOG.md b/CHANGELOG.md index 30a210f..380232a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## [Unreleased] + +- **Breaking:** renamed the env var `PIPELEX_API_URL` to `PIPELEX_BASE_URL` for consistency with the SDK's `base_url` naming. There is no read alias — update your `.env` / environment. + ## [v0.10.0] - 2026-07-01 - **Breaking:** run methods through the hosted Pipelex API instead of the local `pipelex` runtime. The `pipelex` package (and its `[tool.uv.sources]` git pin) is dropped; the starter now depends on `pipelex-sdk` (`PipelexAPIClient`) and `python-dotenv`. diff --git a/CLAUDE.md b/CLAUDE.md index 8a714d3..faf1a4d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -34,7 +34,7 @@ Run specific tests (local only): `make tp TEST=test_function_name` This starter calls the **hosted Pipelex API** via the `pipelex-sdk` package (`PipelexAPIClient`) — it does **not** run Pipelex as a local library. The `.mthds` bundle is read from disk and sent to the API as content (`mthds_contents`); the API runs the method and returns the output. -- Credentials/endpoint come from `PIPELEX_API_URL` / `PIPELEX_API_KEY` (see `.env.example`). `python-dotenv` loads `.env` when running the CLI or tests. +- Credentials/endpoint come from `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` (see `.env.example`). `python-dotenv` loads `.env` when running the CLI or tests. - `my_project/hello_world.py` uses `client.start_and_wait(...)` — the durable start-and-poll path (survives the hosted gateway's ~30s cap, self-heals to blocking `execute` on a bare runner). - Output is loosely-typed JSON: hosted runs carry `main_stuff`; the bare-runner fallback carries `pipe_output`. `find_main_content()` normalizes both. diff --git a/README.md b/README.md index 929d177..fd3adeb 100644 --- a/README.md +++ b/README.md @@ -21,14 +21,14 @@ Once you've created your repository from it, clone it and follow the instruction Access to a **Pipelex API** server. You have two options: -- **Hosted** — currently in private beta. Join the waitlist at [go.pipelex.com/waitlist](https://go.pipelex.com/waitlist). Once you have access, get an API key at [app.pipelex.com](https://app.pipelex.com) and point `PIPELEX_API_URL` at `https://api.pipelex.com` (the default). -- **Self-hosted** — the Pipelex API is open source at [github.com/Pipelex/pipelex-api](https://github.com/Pipelex/pipelex-api). Run it locally or on your own infra and point `PIPELEX_API_URL` at your instance (e.g. `http://127.0.0.1:8081`). +- **Hosted** — currently in private beta. Join the waitlist at [go.pipelex.com/waitlist](https://go.pipelex.com/waitlist). Once you have access, get an API key at [app.pipelex.com](https://app.pipelex.com) and point `PIPELEX_BASE_URL` at `https://api.pipelex.com` (the default). +- **Self-hosted** — the Pipelex API is open source at [github.com/Pipelex/pipelex-api](https://github.com/Pipelex/pipelex-api). Run it locally or on your own infra and point `PIPELEX_BASE_URL` at your instance (e.g. `http://127.0.0.1:8081`). ## Quick start ```bash cp .env.example .env -# edit .env and set PIPELEX_API_KEY (and PIPELEX_API_URL if self-hosting) +# edit .env and set PIPELEX_API_KEY (and PIPELEX_BASE_URL if self-hosting) make install # create the venv and install deps with uv python -m my_project.hello_world # run the hello_world example against the API @@ -45,7 +45,7 @@ my_project/ tests/ integration/ # offline boot/bundle checks + API validate (pipelex_api) e2e/ # full run against the API (inference) -.env.example # PIPELEX_API_URL + PIPELEX_API_KEY +.env.example # PIPELEX_BASE_URL + PIPELEX_API_KEY ``` ## How it works @@ -53,7 +53,7 @@ tests/ `my_project/hello_world.py`: 1. Reads the `.mthds` bundle from disk (`BUNDLE_PATH`). -2. Constructs a `PipelexAPIClient`, which reads `PIPELEX_API_URL` / `PIPELEX_API_KEY` from the environment. +2. Constructs a `PipelexAPIClient`, which reads `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` from the environment. 3. Calls `start_and_wait(pipe_code="hello_world", mthds_contents=[bundle])` — the durable start-and-poll path that survives the hosted gateway's ~30s synchronous cap and self-heals to a blocking `execute` against a bare self-hosted runner. 4. Reads the main output's content (`{"text": ...}`) out of the loosely-typed result and prints it. diff --git a/my_project/hello_world.py b/my_project/hello_world.py index f399c0e..3313039 100644 --- a/my_project/hello_world.py +++ b/my_project/hello_world.py @@ -46,15 +46,15 @@ def find_main_content(results: RunResults) -> dict[str, Any] | None: async def hello_world() -> None: """Run a super-simple Pipelex pipeline through the hosted Pipelex API and print its output. - The `PipelexAPIClient` reads `PIPELEX_API_URL` / `PIPELEX_API_KEY` from the + The `PipelexAPIClient` reads `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` from the environment. `start_and_wait` submits the run and polls it to completion — the durable path that survives the hosted gateway's ~30s synchronous cap, and self-heals to a blocking `execute` against a bare self-hosted runner. """ bundle = BUNDLE_PATH.read_text() - async with PipelexAPIClient() as client: - results = await client.start_and_wait( + async with PipelexAPIClient() as pipelex_client: + results = await pipelex_client.start_and_wait( pipe_code="hello_world", mthds_contents=[bundle], ) @@ -72,6 +72,6 @@ async def hello_world() -> None: if __name__ == "__main__": - # Load .env so PIPELEX_API_URL / PIPELEX_API_KEY are available when run directly. + # Load .env so PIPELEX_BASE_URL / PIPELEX_API_KEY are available when run directly. load_dotenv() asyncio.run(hello_world()) diff --git a/pyproject.toml b/pyproject.toml index 88585aa..fa59528 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ classifiers = [ ] dependencies = [ - "pipelex-sdk==0.1.1", + "pipelex-sdk==0.2.0", "python-dotenv>=1.0.0", ] diff --git a/tests/conftest.py b/tests/conftest.py index 3eb5338..1b5a542 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,5 @@ from dotenv import load_dotenv -# Load .env so the Pipelex API client picks up PIPELEX_API_URL / PIPELEX_API_KEY +# Load .env so the Pipelex API client picks up PIPELEX_BASE_URL / PIPELEX_API_KEY # (the tests marked `pipelex_api` / `inference` reach the hosted API). load_dotenv() diff --git a/tests/integration/test_fundamentals.py b/tests/integration/test_fundamentals.py index 7d30f80..21f1013 100644 --- a/tests/integration/test_fundamentals.py +++ b/tests/integration/test_fundamentals.py @@ -7,9 +7,9 @@ class TestFundamentals: def test_boot(self): # Constructing the client resolves credentials + base URL (no network). - # This fails if PIPELEX_API_URL is malformed. + # This fails if PIPELEX_BASE_URL is malformed. client = PipelexAPIClient() - assert client.api_base_url + assert client.base_url def test_bundle_exists(self): assert BUNDLE_PATH.exists() diff --git a/uv.lock b/uv.lock index 6287946..ddeaf33 100644 --- a/uv.lock +++ b/uv.lock @@ -208,7 +208,7 @@ wheels = [ [[package]] name = "mthds" -version = "0.6.1" +version = "0.7.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" }, @@ -219,9 +219,9 @@ dependencies = [ { name = "tomlkit" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/a7/33/c274de1115b6cbe8ac3475327a622893070f1ca40d9f2d0172d6fcae62b9/mthds-0.6.1.tar.gz", hash = "sha256:3d6b93306708f0ef8971285cfab4d9a488fca364e5db7e63a7dba2d065070eef", size = 132504, upload-time = "2026-07-01T07:09:50.568Z" } +sdist = { url = "https://files.pythonhosted.org/packages/8d/4c/f3ba4c7e0168facb250efd6b8510a5528db8509701591d05b996023322a0/mthds-0.7.1.tar.gz", hash = "sha256:60b8a14770342d67cd63c97cf73eedb14e85f1758bb88c590f0104234c56a8bd", size = 136752, upload-time = "2026-07-02T22:00:09.4Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/ff/52/0538d1bd11067f60f86ad74dc4d36bcad81ff3b549f71b3290aebd3a9aeb/mthds-0.6.1-py3-none-any.whl", hash = "sha256:5cd39d2b6fc5b43c5967162d80178b050748502515eda26916c2b80e8f3f1f42", size = 57705, upload-time = "2026-07-01T07:09:49.256Z" }, + { url = "https://files.pythonhosted.org/packages/29/30/67134348798d1dcc0270009bec3a6a331655a1dd6be3b8c5579190ed4ab2/mthds-0.7.1-py3-none-any.whl", hash = "sha256:b463eaeef67df6e52d70905c6318135ebe950c8eba6a979806dd2d132e4db191", size = 58015, upload-time = "2026-07-02T22:00:07.982Z" }, ] [[package]] @@ -247,7 +247,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.11.2" }, - { name = "pipelex-sdk", specifier = "==0.1.1" }, + { name = "pipelex-sdk", specifier = "==0.2.0" }, { name = "pipelex-tools", marker = "extra == 'dev'", specifier = ">=0.3.2" }, { name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.410" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=9.0.1" }, @@ -342,7 +342,7 @@ wheels = [ [[package]] name = "pipelex-sdk" -version = "0.1.1" +version = "0.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" }, @@ -351,9 +351,9 @@ dependencies = [ { name = "pydantic" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/1e/32/4283ffac0148697f6474ee88362e33cd9954da0f0bd0961bf40cac98a6b0/pipelex_sdk-0.1.1.tar.gz", hash = "sha256:d78d585e4457a1dd0501db56f8b4976d61b7fba14eead8e70a40296167d18787", size = 114643, upload-time = "2026-07-01T07:54:36.268Z" } +sdist = { url = "https://files.pythonhosted.org/packages/53/48/585dd15cb559fe3a83b39eb5279ecca99446f388777a1d97d1823606c0d3/pipelex_sdk-0.2.0.tar.gz", hash = "sha256:2df844f6e79c5e84465855452b0838f29db1d4356897fc6b08b057c837364539", size = 116730, upload-time = "2026-07-02T22:12:07.891Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/45/8d/9b00c0b38a09d0486596a8118caaac0ffb42dc8141215cedf310876b4bbb/pipelex_sdk-0.1.1-py3-none-any.whl", hash = "sha256:e7df4902bc2a65d67efe9edbdfc07672365351ea10a836dd83c28b80a78f0cc0", size = 30768, upload-time = "2026-07-01T07:54:34.776Z" }, + { url = "https://files.pythonhosted.org/packages/dd/31/304cd4cbcb70499563a3dd34f1120bad6c337f314c45df674b5caeb611f7/pipelex_sdk-0.2.0-py3-none-any.whl", hash = "sha256:ea94d40cfcf56fe234294d6324c4ac626c1658c61f3a15d698b968ba9fee2e0d", size = 31459, upload-time = "2026-07-02T22:12:06.689Z" }, ] [[package]] From 1fffaeb0164dcb1c1042d8beaee604df3b9dcd74 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Fri, 3 Jul 2026 01:28:27 +0200 Subject: [PATCH 2/8] feat: Introduce entity extraction example and CLI integration - Added `extract_entities.py` example for extracting people, organizations, and dates from text. - Created `main.mthds` for the entity extraction method. - Implemented `find_main_content` in `run_output.py` to normalize API output. - Developed `runner.py` for executing pipelines in blocking and durable modes. - Updated `pyproject.toml` to include new dependencies and CLI entry point. - Removed obsolete `hello_world` files and tests. - Added unit and integration tests for the new entity extraction functionality. - Enhanced error handling and validation in the new implementation. --- my_project/cli.py | 171 ++++++++++++++++++ my_project/errors.py | 86 +++++++++ my_project/examples/__init__.py | 0 my_project/examples/extract_entities.py | 39 ++++ my_project/hello_world.mthds | 14 -- my_project/hello_world.py | 77 -------- .../methods/extract-entities/main.mthds | 30 +++ my_project/run_output.py | 62 +++++++ my_project/runner.py | 104 +++++++++++ pyproject.toml | 11 +- tests/e2e/test_my_project.py | 26 ++- tests/integration/test_fundamentals.py | 2 +- tests/unit/test_cli.py | 75 ++++++++ tests/unit/test_errors.py | 61 +++++++ tests/unit/test_extract_entities.py | 40 ++++ tests/unit/test_hello_world.py | 53 ------ tests/unit/test_run_output.py | 53 ++++++ uv.lock | 85 +++++++++ 18 files changed, 838 insertions(+), 151 deletions(-) create mode 100644 my_project/cli.py create mode 100644 my_project/errors.py create mode 100644 my_project/examples/__init__.py create mode 100644 my_project/examples/extract_entities.py delete mode 100644 my_project/hello_world.mthds delete mode 100644 my_project/hello_world.py create mode 100644 my_project/methods/extract-entities/main.mthds create mode 100644 my_project/run_output.py create mode 100644 my_project/runner.py create mode 100644 tests/unit/test_cli.py create mode 100644 tests/unit/test_errors.py create mode 100644 tests/unit/test_extract_entities.py delete mode 100644 tests/unit/test_hello_world.py create mode 100644 tests/unit/test_run_output.py diff --git a/my_project/cli.py b/my_project/cli.py new file mode 100644 index 0000000..e74b597 --- /dev/null +++ b/my_project/cli.py @@ -0,0 +1,171 @@ +"""The `my-project` CLI — demo Pipelex methods behind a Typer app. + +Commands stay thin: parse arguments, dispatch on execution mode via +`my_project.runner`, narrow + render via the matching `my_project.examples` +module. SDK errors are caught once per command (in `_run_cli`) and presented by +`my_project.errors`; anything unexpected crashes loudly. +""" + +import asyncio +from pathlib import Path +from typing import Annotated, Any, Coroutine, TypeVar + +import httpx +import typer +from dotenv import load_dotenv +from mthds.protocol.exceptions import PipelineRequestError +from pipelex_sdk.runs import RunResultCompleted, RunResultFailed, RunResultRunning, RunResults, RunResultState +from rich.console import Console + +from my_project.errors import present_error +from my_project.examples import extract_entities as extract_entities_example +from my_project.run_output import find_main_content +from my_project.runner import ( + ExecutionMode, + fetch_run_result, + fetch_run_status, + progress_console, + run_blocking, + run_durable_attended, + start_detached, + wait_for_run, +) + +ResultT = TypeVar("ResultT") + +app = typer.Typer(no_args_is_help=True, help="Run the demo Pipelex methods through the Pipelex API.") +runs_app = typer.Typer(no_args_is_help=True, help="Inspect and resume durable runs by id.") +app.add_typer(runs_app, name="runs") + +# Results go to stdout (pipeable); progress/status chatter goes to stderr (see runner.py). +output_console = Console() + +MODE_HELP = "How to execute the run: `durable` (start + poll, survives anything) or `blocking` (single call, ~30s cap on hosted)." +DETACH_HELP = "Start the run and exit immediately; fetch it later with `my-project runs ...` (durable mode only)." + + +@app.callback() +def main() -> None: + """Load .env so PIPELEX_BASE_URL / PIPELEX_API_KEY are available.""" + load_dotenv() + + +@app.command(name="extract-entities") +def extract_entities( + text: Annotated[str | None, typer.Argument(help="The text to extract entities from.")] = None, + file: Annotated[Path | None, typer.Option("--file", help="Read the input text from a file instead of the argument.")] = None, + mode: Annotated[ExecutionMode, typer.Option(envvar="PIPELEX_EXECUTION_MODE", help=MODE_HELP)] = ExecutionMode.DURABLE, + detach: Annotated[bool, typer.Option("--detach", help=DETACH_HELP)] = False, +) -> None: + """Extract people, organizations, and dates from a piece of text.""" + input_text = _read_text_input(text=text, file=file) + bundle = extract_entities_example.BUNDLE_PATH.read_text() + results = _dispatch( + pipe_code=extract_entities_example.PIPE_CODE, + bundle=bundle, + inputs={"text": input_text}, + mode=mode, + detach=detach, + ) + if results is None: + return + # Narrow into the typed model (validates the concept's shape), then print it + # as JSON — the same rendering `runs result` / `runs wait` give. + entities = extract_entities_example.parse(results) + output_console.print_json(data=entities.model_dump()) + + +@runs_app.command(name="status") +def runs_status(run_id: Annotated[str, typer.Argument(help="The pipeline run id printed when the run started.")]) -> None: + """Show a run's coarse status without waiting.""" + run = _run_cli(fetch_run_status(run_id)) + pipe_part = f" (pipe: {run.pipe_code})" if run.pipe_code else "" + output_console.print(f"{run.pipeline_run_id}: [bold]{run.status}[/bold]{pipe_part}") + if run.degraded: + output_console.print("[yellow]Status is degraded — last-known value, the status backend was unreachable; retry shortly.[/yellow]") + + +@runs_app.command(name="result") +def runs_result(run_id: Annotated[str, typer.Argument(help="The pipeline run id printed when the run started.")]) -> None: + """Fetch a run's result if it is finished (no waiting).""" + state = _run_cli(fetch_run_result(run_id)) + _render_result_state(state) + + +@runs_app.command(name="wait") +def runs_wait(run_id: Annotated[str, typer.Argument(help="The pipeline run id printed when the run started.")]) -> None: + """Poll a run to completion, then print its raw result.""" + results = _run_cli(wait_for_run(run_id)) + _print_raw_results(results) + + +def _read_text_input(*, text: str | None, file: Path | None) -> str: + if text is not None and file is not None: + msg = "Give the text either as an argument or via --file, not both." + raise typer.BadParameter(msg) + if file is not None: + return file.read_text() + if text is not None: + return text + msg = "Give the text to process as an argument, or point --file at a text file." + raise typer.BadParameter(msg) + + +def _dispatch(*, pipe_code: str, bundle: str, inputs: dict[str, Any], mode: ExecutionMode, detach: bool) -> RunResults | None: + """Run the pipe in the requested mode; returns None when detached (id already printed).""" + if detach: + match mode: + case ExecutionMode.BLOCKING: + msg = "--detach starts a durable run; it cannot be combined with --mode blocking." + raise typer.BadParameter(msg) + case ExecutionMode.DURABLE: + pass + run_id = _run_cli(start_detached(pipe_code=pipe_code, bundle=bundle, inputs=inputs)) + print(run_id) + progress_console.print(f"Run started — fetch it later with: [bold]my-project runs wait {run_id}[/bold]") + return None + match mode: + case ExecutionMode.BLOCKING: + return _run_cli(run_blocking(pipe_code=pipe_code, bundle=bundle, inputs=inputs)) + case ExecutionMode.DURABLE: + return _run_cli(run_durable_attended(pipe_code=pipe_code, bundle=bundle, inputs=inputs)) + + +def _run_cli(coro: Coroutine[Any, Any, ResultT]) -> ResultT: + """Await a runner coroutine, presenting SDK errors and Ctrl-C as clean exits.""" + try: + return asyncio.run(coro) + except (PipelineRequestError, httpx.HTTPStatusError) as exc: + presentation = present_error(exc) + progress_console.print(f"[red]Error:[/red] {presentation.message}") + if presentation.hint: + progress_console.print(f"[yellow]Hint:[/yellow] {presentation.hint}") + raise typer.Exit(1) from exc + except KeyboardInterrupt as exc: + # The resume hint was already printed by the runner; the run keeps executing server-side. + raise typer.Exit(130) from exc + + +def _render_result_state(state: RunResultState) -> None: + match state: + case RunResultRunning(): + progress_console.print( + f"Run {state.pipeline_run_id} is still running — wait for it with: [bold]my-project runs wait {state.pipeline_run_id}[/bold]" + ) + case RunResultCompleted(): + _print_raw_results(state.result) + case RunResultFailed(): + progress_console.print(f"[red]Run {state.pipeline_run_id} ended with status {state.status}: {state.message}[/red]") + raise typer.Exit(1) + + +def _print_raw_results(results: RunResults) -> None: + """Print the run's main content as JSON — generic, no per-example narrowing.""" + content: Any = find_main_content(results) + if content is None: + content = results.main_stuff if results.main_stuff is not None else results.pipe_output + output_console.print_json(data=content) + + +if __name__ == "__main__": + app() diff --git a/my_project/errors.py b/my_project/errors.py new file mode 100644 index 0000000..e94785b --- /dev/null +++ b/my_project/errors.py @@ -0,0 +1,86 @@ +"""Present SDK errors as actionable CLI messages. + +This module defines no exception classes — it is a presentation mapper. Each +CLI command catches `PipelineRequestError` (the base of every error the +`pipelex-sdk` client raises) exactly once at its root, turns it into a +`(message, hint)` pair here, and exits non-zero. Unexpected exceptions are +deliberately NOT caught anywhere: they crash loudly with a full traceback. +""" + +from typing import NamedTuple + +import httpx +from mthds.protocol.exceptions import PipelineRequestError +from pipelex_sdk.errors import ( + ApiResponseError, + ApiUnreachableError, + PipelineExecuteTimeoutError, + RunFailedError, + RunLifecycleUnavailableError, + RunTimeoutError, +) + + +class ErrorPresentation(NamedTuple): + """What the CLI shows for a failed command: the error and what to do about it.""" + + message: str + hint: str | None + + +def present_error(exc: PipelineRequestError | httpx.HTTPStatusError) -> ErrorPresentation: + """Map an SDK error to a CLI-facing message and an actionable hint. + + The SDK's protocol routes (`execute`, `start`, `runs/*`) surface non-2xx + responses as raw `httpx.HTTPStatusError` (the inherited regime); the typed + `ApiResponseError` only rides the product routes. Both are mapped here so + an auth failure gets the API-key hint whichever route raised it. + """ + if isinstance(exc, httpx.HTTPStatusError): + return _present_http_status_error(exc) + if isinstance(exc, PipelineExecuteTimeoutError): + return ErrorPresentation( + message=f"The blocking run exceeded the hosted gateway's ~30s synchronous cap ({exc.elapsed_seconds:.0f}s elapsed).", + hint="Retry with `--mode durable` — the durable path survives long runs.", + ) + if isinstance(exc, RunLifecycleUnavailableError): + return ErrorPresentation( + message=f"The server at {exc.api_url} has no run store (durable run lifecycle unavailable).", + hint="You are talking to a bare runner — retry with `--mode blocking`.", + ) + if isinstance(exc, ApiResponseError): + if exc.status in (401, 403): + return ErrorPresentation( + message=f"The API rejected the request ({exc.status} {exc.status_text}).", + hint="Set PIPELEX_API_KEY in your environment or .env file — get a key at https://app.pipelex.com", + ) + return ErrorPresentation( + message=f"The API answered {exc.status} {exc.status_text}: {exc.server_message or exc}", + hint=None, + ) + if isinstance(exc, ApiUnreachableError): + return ErrorPresentation( + message=f"Could not reach the Pipelex API at {exc.api_url}.", + hint="Check PIPELEX_BASE_URL — and if you self-host, make sure your runner is up.", + ) + if isinstance(exc, RunFailedError): + return ErrorPresentation( + message=f"Run {exc.run_id} ended with status {exc.status}: {exc}", + hint=f"Inspect it with `my-project runs status {exc.run_id}`.", + ) + if isinstance(exc, RunTimeoutError): + return ErrorPresentation( + message=f"Gave up waiting for run {exc.run_id} after {exc.timeout_seconds:.0f}s — the run is still executing server-side.", + hint=f"Resume waiting with `my-project runs wait {exc.run_id}`.", + ) + return ErrorPresentation(message=str(exc), hint=None) + + +def _present_http_status_error(exc: httpx.HTTPStatusError) -> ErrorPresentation: + status_code = exc.response.status_code + if status_code in (401, 403): + return ErrorPresentation( + message=f"The API rejected the request ({status_code} {exc.response.reason_phrase}).", + hint="Set PIPELEX_API_KEY in your environment or .env file — get a key at https://app.pipelex.com", + ) + return ErrorPresentation(message=str(exc), hint=None) diff --git a/my_project/examples/__init__.py b/my_project/examples/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/my_project/examples/extract_entities.py b/my_project/examples/extract_entities.py new file mode 100644 index 0000000..5e9abe4 --- /dev/null +++ b/my_project/examples/extract_entities.py @@ -0,0 +1,39 @@ +"""Example: extract people, organizations, and dates from a piece of text. + +This module is the "copy me" unit for swapping in your own pipeline: a bundle +path, a Pydantic model mirroring the output concept, and a `parse` narrower +that turns the loose run result into that model. The CLI prints the parsed +model as JSON — same rendering as `runs result` / `runs wait`. +""" + +from pathlib import Path + +from pipelex_sdk.runs import RunResults +from pydantic import BaseModel + +from my_project.run_output import find_main_content + +BUNDLE_PATH = Path(__file__).parent.parent / "methods" / "extract-entities" / "main.mthds" +PIPE_CODE = "extract_entities" + + +class ExtractedEntities(BaseModel): + """Mirror of the bundle's `ExtractedEntities` output concept.""" + + people: list[str] + orgs: list[str] + dates: list[str] + + +def parse(results: RunResults) -> ExtractedEntities: + """Narrow a run result into a typed `ExtractedEntities`. + + Raises: + RuntimeError: The run produced no output content at all. + pydantic.ValidationError: The content doesn't match the concept's shape. + """ + content = find_main_content(results) + if content is None: + msg = "The run returned no output content." + raise RuntimeError(msg) + return ExtractedEntities.model_validate(content) diff --git a/my_project/hello_world.mthds b/my_project/hello_world.mthds deleted file mode 100644 index 26c1d52..0000000 --- a/my_project/hello_world.mthds +++ /dev/null @@ -1,14 +0,0 @@ - - -domain = "quick_start" -description = "Discovering Pipelex" - -[pipe] -[pipe.hello_world] -type = "PipeLLM" -description = "Write text about Hello World." -output = "Text" -model = { model = "gpt-4o-mini", temperature = 0.9, max_tokens = "auto" } -prompt = """ -Write a haiku about Hello World. -""" diff --git a/my_project/hello_world.py b/my_project/hello_world.py deleted file mode 100644 index 3313039..0000000 --- a/my_project/hello_world.py +++ /dev/null @@ -1,77 +0,0 @@ -import asyncio -from pathlib import Path -from typing import Any, cast - -from dotenv import load_dotenv -from pipelex_sdk.client import PipelexAPIClient -from pipelex_sdk.runs import RunResults - -# The .mthds bundle lives next to this module. -BUNDLE_PATH = Path(__file__).parent / "hello_world.mthds" - - -def find_main_content(results: RunResults) -> dict[str, Any] | None: - """Read the main output's content dict out of a run result. - - The Pipelex API returns one of two shapes (both opaque JSON), so we - normalize both here: - - Hosted runs carry `main_stuff` — the main output's content directly - (for our `hello_world` pipe: `{"text": "..."}`). - - The bare-runner blocking fallback carries `pipe_output` - (`{"working_memory": {"root": {: {"content": ...}}}}`); we return - the first entry's content. - """ - main_stuff: Any = results.main_stuff - if isinstance(main_stuff, dict): - return cast("dict[str, Any]", main_stuff) - - pipe_output = results.pipe_output - if pipe_output is None: - return None - working_memory = pipe_output.get("working_memory") - if not isinstance(working_memory, dict): - return None - root = cast("dict[str, Any]", working_memory).get("root") - if not isinstance(root, dict): - return None - for entry in cast("dict[str, Any]", root).values(): - if not isinstance(entry, dict): - continue - content = cast("dict[str, Any]", entry).get("content") - if isinstance(content, dict): - return cast("dict[str, Any]", content) - return None - - -async def hello_world() -> None: - """Run a super-simple Pipelex pipeline through the hosted Pipelex API and print its output. - - The `PipelexAPIClient` reads `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` from the - environment. `start_and_wait` submits the run and polls it to completion — the - durable path that survives the hosted gateway's ~30s synchronous cap, and - self-heals to a blocking `execute` against a bare self-hosted runner. - """ - bundle = BUNDLE_PATH.read_text() - - async with PipelexAPIClient() as pipelex_client: - results = await pipelex_client.start_and_wait( - pipe_code="hello_world", - mthds_contents=[bundle], - ) - - content = find_main_content(results) - if content is None: - raise RuntimeError("The pipeline returned no output content.") - - generated_text = content.get("text") - if not isinstance(generated_text, str) or not generated_text: - raise RuntimeError("The pipeline returned no text output.") - - print("Your first Pipelex output:\n") - print(generated_text) - - -if __name__ == "__main__": - # Load .env so PIPELEX_BASE_URL / PIPELEX_API_KEY are available when run directly. - load_dotenv() - asyncio.run(hello_world()) diff --git a/my_project/methods/extract-entities/main.mthds b/my_project/methods/extract-entities/main.mthds new file mode 100644 index 0000000..ebcbbc6 --- /dev/null +++ b/my_project/methods/extract-entities/main.mthds @@ -0,0 +1,30 @@ +domain = "extract_entities" +description = "Extract people, organizations, and dates from a piece of text." +main_pipe = "extract_entities" + +[concept.ExtractedEntities] +description = "Named entities extracted from a piece of text." + +[concept.ExtractedEntities.structure] +people = { type = "list", description = "Names of people mentioned in the text", required = true } +orgs = { type = "list", description = "Names of organizations mentioned in the text", required = true } +dates = { type = "list", description = "Dates or time references mentioned in the text", required = true } + +[pipe.extract_entities] +type = "PipeLLM" +description = "Extract people, organizations, and dates from the input text." +inputs = { text = "Text" } +output = "ExtractedEntities" +prompt = """ +Extract named entities from the following text. + +Return three lists: +- people: full names of individual people mentioned. +- orgs: organizations, companies, agencies, or institutions mentioned. +- dates: dates or explicit time references (e.g. "March 5th, 2026", "last Tuesday"). + +If a category has no matches, return an empty list. + +Text: +@text +""" diff --git a/my_project/run_output.py b/my_project/run_output.py new file mode 100644 index 0000000..f9e7a6d --- /dev/null +++ b/my_project/run_output.py @@ -0,0 +1,62 @@ +"""Normalize the Pipelex API's run-output shapes into plain content dicts. + +The API returns one of two opaque JSON shapes depending on the path taken: +hosted durable runs carry `main_stuff` (the main output's content directly), +while the blocking / bare-runner path carries `pipe_output` (a serialized +working memory). Everything downstream (the per-example `parse` functions) +goes through `find_main_content` so it never has to care which path ran. +""" + +from typing import Any, cast + +from mthds.runners.api.models import DictRunResultExecute +from pipelex_sdk.runs import RunResults + + +def find_main_content(results: RunResults) -> dict[str, Any] | None: + """Read the main output's content dict out of a run result. + + The Pipelex API returns one of two shapes (both opaque JSON), so we + normalize both here: + - Hosted runs carry `main_stuff` — the main output's content directly + (e.g. `{"people": [...], "orgs": [...], "dates": [...]}`). + - The bare-runner blocking fallback carries `pipe_output` + (`{"working_memory": {"root": {: {"content": ...}}}}`); we return + the first entry's content. + """ + main_stuff: Any = results.main_stuff + if isinstance(main_stuff, dict): + return cast("dict[str, Any]", main_stuff) + + pipe_output = results.pipe_output + if pipe_output is None: + return None + working_memory = pipe_output.get("working_memory") + if not isinstance(working_memory, dict): + return None + root = cast("dict[str, Any]", working_memory).get("root") + if not isinstance(root, dict): + return None + for entry in cast("dict[str, Any]", root).values(): + if not isinstance(entry, dict): + continue + content = cast("dict[str, Any]", entry).get("content") + if isinstance(content, dict): + return cast("dict[str, Any]", content) + return None + + +def to_run_results(result: DictRunResultExecute) -> RunResults: + """Normalize a blocking `execute` response into the lifecycle's `RunResults`. + + The blocking path returns the runner's native shape (`pipe_output`); the + hosted-durable artifacts (`main_stuff`, `graph_spec`) don't exist on that + path and stay `None`. This mirrors what the SDK's `start_and_wait` does + internally, so both execution modes hand the same type to `find_main_content`. + """ + return RunResults( + pipeline_run_id=result.pipeline_run_id, + main_stuff=None, + graph_spec=None, + pipe_output=result.pipe_output.model_dump(), + ) diff --git a/my_project/runner.py b/my_project/runner.py new file mode 100644 index 0000000..4e69bd0 --- /dev/null +++ b/my_project/runner.py @@ -0,0 +1,104 @@ +"""Execution-mode dispatch for the CLI: blocking, durable attended, durable detached. + +Each function opens its own `PipelexAPIClient` (credentials come from +`PIPELEX_API_KEY` / `PIPELEX_BASE_URL`), demonstrates exactly one SDK lifecycle +call, and renders progress with Rich on stderr — stdout stays clean for results. + +- blocking -> `client.execute` (one call, dies at the hosted ~30s cap) +- durable attended -> `client.start` + `client.wait_for_result` (survives anything) +- durable detached -> `client.start` only (come back later with `runs ...`) + +The SDK also offers `start_and_wait`, a self-healing one-liner that picks the +right path by itself — this starter branches explicitly because teaching the +difference is the point. +""" + +import asyncio +from enum import Enum +from typing import Any + +from pipelex_sdk.client import PipelexAPIClient +from pipelex_sdk.runs import PollInfo, RunRead, RunResults, RunResultState, WaitForResultOptions +from rich.console import Console + +from my_project.run_output import to_run_results + +# Progress and lifecycle chatter go to stderr so stdout stays pipeable. +progress_console = Console(stderr=True) + + +class ExecutionMode(str, Enum): + """How a run is executed against the API — see the module docstring.""" + + BLOCKING = "blocking" + DURABLE = "durable" + + +async def run_blocking(*, pipe_code: str, bundle: str, inputs: dict[str, Any] | None = None) -> RunResults: + """Execute synchronously (`POST /v1/execute`) and wait for the response. + + Simple, but behind the hosted gateway a run longer than ~30s raises + `PipelineExecuteTimeoutError` — the CLI turns that into a "use durable" hint. + """ + async with PipelexAPIClient() as client: + with progress_console.status("Running (blocking)…"): + result = await client.execute(pipe_code=pipe_code, mthds_contents=[bundle], inputs=inputs) + return to_run_results(result) + + +async def run_durable_attended(*, pipe_code: str, bundle: str, inputs: dict[str, Any] | None = None) -> RunResults: + """Start a durable run, print its id immediately, then poll it to completion. + + The id is printed before polling so the run is never lost: Ctrl-C leaves it + executing server-side and you can resume with `my-project runs wait `. + """ + async with PipelexAPIClient() as client: + start_result = await client.start(pipe_code=pipe_code, mthds_contents=[bundle], inputs=inputs) + run_id = start_result.pipeline_run_id + progress_console.print(f"Run started: [bold]{run_id}[/bold]") + return await _attend(client=client, run_id=run_id) + + +async def start_detached(*, pipe_code: str, bundle: str, inputs: dict[str, Any] | None = None) -> str: + """Start a durable run and return its id without waiting. + + The run keeps executing server-side; fetch it later with + `my-project runs status|result|wait ` — even from another terminal. + """ + async with PipelexAPIClient() as client: + start_result = await client.start(pipe_code=pipe_code, mthds_contents=[bundle], inputs=inputs) + return start_result.pipeline_run_id + + +async def wait_for_run(run_id: str) -> RunResults: + """Poll an already-started run to completion with a live status line.""" + async with PipelexAPIClient() as client: + return await _attend(client=client, run_id=run_id) + + +async def fetch_run_status(run_id: str) -> RunRead: + """Fetch a run's coarse status (`GET /v1/runs/{id}/status`).""" + async with PipelexAPIClient() as client: + return await client.get_run_status(run_id) + + +async def fetch_run_result(run_id: str) -> RunResultState: + """Fetch a run's result state (`GET /v1/runs/{id}/results`) without polling.""" + async with PipelexAPIClient() as client: + return await client.get_run_result(run_id) + + +async def _attend(*, client: PipelexAPIClient, run_id: str) -> RunResults: + """Poll a run to its terminal state, driving a Rich status line per poll.""" + short_id = run_id[:8] + with progress_console.status(f"Run {short_id}… in progress") as status: + + def on_poll(info: PollInfo) -> None: + status.update(f"Run {short_id}… in progress — {info.elapsed_seconds:.0f}s, poll #{info.attempt}") + + try: + return await client.wait_for_result(run_id, options=WaitForResultOptions(on_poll=on_poll)) + except asyncio.CancelledError: + # Ctrl-C: the run keeps executing server-side — tell the user how to pick it back up. + progress_console.print(f"\nInterrupted — the run is still executing. Resume with: [bold]my-project runs wait {run_id}[/bold]") + raise diff --git a/pyproject.toml b/pyproject.toml index fa59528..7e6d813 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,20 +17,29 @@ classifiers = [ ] dependencies = [ + "httpx>=0.27.0", "pipelex-sdk==0.2.0", "python-dotenv>=1.0.0", + "typer>=0.15.0", ] +[project.scripts] +my-project = "my_project.cli:app" + [tool.setuptools] -packages = ["my_project"] +packages = ["my_project", "my_project.examples"] include-package-data = true +[tool.setuptools.package-data] +my_project = ["py.typed", "methods/*/main.mthds"] + [project.optional-dependencies] dev = [ "mypy>=1.11.2", "pipelex-tools>=0.3.2", "pyright>=1.1.410", "pytest>=9.0.1", + "pytest-mock>=3.14.0", "pytest-sugar>=1.0.0", "pytest_asyncio>=0.24.0", "ruff>=0.6.8", diff --git a/tests/e2e/test_my_project.py b/tests/e2e/test_my_project.py index 55b6bea..199a092 100644 --- a/tests/e2e/test_my_project.py +++ b/tests/e2e/test_my_project.py @@ -1,11 +1,27 @@ import pytest -from my_project.hello_world import hello_world +from my_project.examples.extract_entities import BUNDLE_PATH, PIPE_CODE, parse +from my_project.runner import run_blocking, run_durable_attended + +SAMPLE_TEXT = ( + "Marie Curie joined the University of Paris in 1906, two years after Pierre Curie won recognition from the Royal Swedish Academy of Sciences." +) @pytest.mark.inference +@pytest.mark.llm @pytest.mark.pipelex_api -class TestMyProject: - async def test_hello_world(self): - # Runs the pipeline end-to-end through the hosted Pipelex API (real inference). - await hello_world() +class TestExtractEntities: + async def test_durable(self): + # Full durable lifecycle through the hosted API: start + poll + narrow. + bundle = BUNDLE_PATH.read_text() + results = await run_durable_attended(pipe_code=PIPE_CODE, bundle=bundle, inputs={"text": SAMPLE_TEXT}) + entities = parse(results) + assert any("Curie" in person for person in entities.people) + + async def test_blocking(self): + # Blocking `execute` path (extraction is fast enough for the ~30s cap). + bundle = BUNDLE_PATH.read_text() + results = await run_blocking(pipe_code=PIPE_CODE, bundle=bundle, inputs={"text": SAMPLE_TEXT}) + entities = parse(results) + assert any("Curie" in person for person in entities.people) diff --git a/tests/integration/test_fundamentals.py b/tests/integration/test_fundamentals.py index 21f1013..2ce6830 100644 --- a/tests/integration/test_fundamentals.py +++ b/tests/integration/test_fundamentals.py @@ -1,7 +1,7 @@ import pytest from pipelex_sdk.client import PipelexAPIClient -from my_project.hello_world import BUNDLE_PATH +from my_project.examples.extract_entities import BUNDLE_PATH class TestFundamentals: diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py new file mode 100644 index 0000000..b7813a7 --- /dev/null +++ b/tests/unit/test_cli.py @@ -0,0 +1,75 @@ +from pathlib import Path + +from pipelex_sdk.runs import RunResults +from pytest_mock import MockerFixture +from typer.testing import CliRunner + +from my_project.cli import app + +ENTITIES_CONTENT = {"people": ["Marie Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} + +runner = CliRunner() + + +class TestCli: + def test_help_lists_commands(self): + result = runner.invoke(app, ["--help"]) + assert result.exit_code == 0 + assert "extract-entities" in result.output + assert "runs" in result.output + + def test_runs_help_lists_subcommands(self): + result = runner.invoke(app, ["runs", "--help"]) + assert result.exit_code == 0 + for subcommand in ("status", "result", "wait"): + assert subcommand in result.output + + def test_extract_entities_requires_input(self): + result = runner.invoke(app, ["extract-entities"]) + assert result.exit_code != 0 + + def test_extract_entities_rejects_both_text_and_file(self, tmp_path: Path): + input_file = tmp_path / "input.txt" + input_file.write_text("from a file") + result = runner.invoke(app, ["extract-entities", "inline text", "--file", str(input_file)]) + assert result.exit_code != 0 + + def test_extract_entities_rejects_bad_mode(self): + result = runner.invoke(app, ["extract-entities", "some text", "--mode", "bogus"]) + assert result.exit_code != 0 + + def test_extract_entities_rejects_blocking_detach(self): + result = runner.invoke(app, ["extract-entities", "some text", "--mode", "blocking", "--detach"]) + assert result.exit_code != 0 + + def test_default_mode_is_durable(self, mocker: MockerFixture): + durable_mock = mocker.patch( + "my_project.cli.run_durable_attended", return_value=RunResults(pipeline_run_id="run-1", main_stuff=ENTITIES_CONTENT) + ) + result = runner.invoke(app, ["extract-entities", "some text"]) + assert result.exit_code == 0 + durable_mock.assert_awaited_once() + assert "Marie Curie" in result.output + + def test_env_var_selects_blocking_mode(self, mocker: MockerFixture): + blocking_mock = mocker.patch("my_project.cli.run_blocking", return_value=RunResults(pipeline_run_id="run-2", main_stuff=ENTITIES_CONTENT)) + result = runner.invoke(app, ["extract-entities", "some text"], env={"PIPELEX_EXECUTION_MODE": "blocking"}) + assert result.exit_code == 0 + blocking_mock.assert_awaited_once() + + def test_file_input_is_read(self, mocker: MockerFixture, tmp_path: Path): + durable_mock = mocker.patch( + "my_project.cli.run_durable_attended", return_value=RunResults(pipeline_run_id="run-3", main_stuff=ENTITIES_CONTENT) + ) + input_file = tmp_path / "input.txt" + input_file.write_text("text from a file") + result = runner.invoke(app, ["extract-entities", "--file", str(input_file)]) + assert result.exit_code == 0 + assert durable_mock.await_args is not None + assert durable_mock.await_args.kwargs["inputs"] == {"text": "text from a file"} + + def test_detach_prints_run_id(self, mocker: MockerFixture): + mocker.patch("my_project.cli.start_detached", return_value="run-abc123") + result = runner.invoke(app, ["extract-entities", "some text", "--detach"]) + assert result.exit_code == 0 + assert "run-abc123" in result.output diff --git a/tests/unit/test_errors.py b/tests/unit/test_errors.py new file mode 100644 index 0000000..684ba27 --- /dev/null +++ b/tests/unit/test_errors.py @@ -0,0 +1,61 @@ +import httpx +from pipelex_sdk.errors import ( + ApiUnreachableError, + PipelineExecuteTimeoutError, + RunFailedError, + RunLifecycleUnavailableError, + RunTimeoutError, +) +from pipelex_sdk.runs import RunStatus + +from my_project.errors import present_error + + +def _http_status_error(status_code: int) -> httpx.HTTPStatusError: + request = httpx.Request("POST", "https://api.pipelex.com/v1/execute") + response = httpx.Response(status_code, request=request) + return httpx.HTTPStatusError("boom", request=request, response=response) + + +class TestPresentError: + def test_execute_timeout_hints_durable(self): + presentation = present_error(PipelineExecuteTimeoutError("timed out", elapsed_seconds=31.2)) + assert "~30s" in presentation.message + assert presentation.hint is not None + assert "--mode durable" in presentation.hint + + def test_lifecycle_unavailable_hints_blocking(self): + presentation = present_error(RunLifecycleUnavailableError("no run store", api_url="http://localhost:8000")) + assert "http://localhost:8000" in presentation.message + assert presentation.hint is not None + assert "--mode blocking" in presentation.hint + + def test_http_auth_error_hints_api_key(self): + # The protocol routes (execute/start/runs) raise raw httpx.HTTPStatusError, + # not ApiResponseError — an auth failure must still get the key hint. + for status_code in (401, 403): + presentation = present_error(_http_status_error(status_code)) + assert str(status_code) in presentation.message + assert presentation.hint is not None + assert "PIPELEX_API_KEY" in presentation.hint + + def test_http_server_error_has_no_hint(self): + presentation = present_error(_http_status_error(500)) + assert presentation.hint is None + + def test_unreachable_hints_base_url(self): + presentation = present_error(ApiUnreachableError("connect failed", api_url="http://nowhere.invalid")) + assert "http://nowhere.invalid" in presentation.message + assert presentation.hint is not None + assert "PIPELEX_BASE_URL" in presentation.hint + + def test_run_failed_names_run_id(self): + presentation = present_error(RunFailedError("run failed", run_id="run-9", status=RunStatus.FAILED)) + assert "run-9" in presentation.message + assert presentation.hint is not None + assert "runs status run-9" in presentation.hint + + def test_run_timeout_hints_wait(self): + presentation = present_error(RunTimeoutError("too slow", run_id="run-9", timeout_seconds=1200.0)) + assert presentation.hint is not None + assert "runs wait run-9" in presentation.hint diff --git a/tests/unit/test_extract_entities.py b/tests/unit/test_extract_entities.py new file mode 100644 index 0000000..e6d9705 --- /dev/null +++ b/tests/unit/test_extract_entities.py @@ -0,0 +1,40 @@ +import pytest +from pipelex_sdk.runs import RunResults +from pydantic import ValidationError + +from my_project.examples.extract_entities import parse + +ENTITIES_CONTENT = {"people": ["Marie Curie", "Pierre Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} + + +class TestExtractEntitiesParse: + def test_parse_main_stuff_shape(self): + results = RunResults(pipeline_run_id="run-1", main_stuff=ENTITIES_CONTENT) + entities = parse(results) + assert entities.people == ["Marie Curie", "Pierre Curie"] + assert entities.orgs == ["University of Paris"] + assert entities.dates == ["1906"] + + def test_parse_pipe_output_shape(self): + results = RunResults( + pipeline_run_id="run-2", + pipe_output={ + "working_memory": { + "root": {"extracted_entities": {"concept": "extract_entities.ExtractedEntities", "content": ENTITIES_CONTENT}}, + "aliases": {}, + }, + "pipeline_run_id": "run-2", + }, + ) + entities = parse(results) + assert entities.people == ["Marie Curie", "Pierre Curie"] + + def test_parse_shape_mismatch_raises(self): + results = RunResults(pipeline_run_id="run-3", main_stuff={"people": "not-a-list", "orgs": [], "dates": []}) + with pytest.raises(ValidationError): + parse(results) + + def test_parse_no_content_raises(self): + results = RunResults(pipeline_run_id="run-4") + with pytest.raises(RuntimeError, match="no output content"): + parse(results) diff --git a/tests/unit/test_hello_world.py b/tests/unit/test_hello_world.py deleted file mode 100644 index b6589e5..0000000 --- a/tests/unit/test_hello_world.py +++ /dev/null @@ -1,53 +0,0 @@ -from typing import Any - -import pytest -from pytest import CaptureFixture, MonkeyPatch - -from my_project import hello_world as hello_world_module -from my_project.hello_world import hello_world - - -class _FakeResults: - # Minimal stand-in for pipelex_sdk.runs.RunResults: find_main_content() - # reads only `.main_stuff` / `.pipe_output`. - def __init__(self, main_stuff: Any) -> None: - self.main_stuff = main_stuff - self.pipe_output = None - - -class _FakeClient: - # Stand-in for PipelexAPIClient used as `async with ... as client`, so no - # network is touched: __aenter__ returns the client and start_and_wait - # yields the canned results. - def __init__(self, results: _FakeResults) -> None: - self._results = results - - async def __aenter__(self) -> "_FakeClient": - return self - - async def __aexit__(self, *_exc_info: object) -> None: - return None - - async def start_and_wait(self, *, pipe_code: str, mthds_contents: list[str]) -> _FakeResults: - return self._results - - -class TestHelloWorldOutput: - def _patch_client(self, monkeypatch: MonkeyPatch, main_stuff: Any) -> None: - results = _FakeResults(main_stuff=main_stuff) - monkeypatch.setattr(hello_world_module, "PipelexAPIClient", lambda: _FakeClient(results)) - - async def test_missing_text_raises(self, monkeypatch: MonkeyPatch): - # A valid content dict without a `text` key must fail loudly instead of - # printing `None` and exiting successfully. - self._patch_client(monkeypatch, main_stuff={"not_text": "oops"}) - - with pytest.raises(RuntimeError, match="no text output"): - await hello_world() - - async def test_valid_text_prints(self, monkeypatch: MonkeyPatch, capsys: CaptureFixture[str]): - self._patch_client(monkeypatch, main_stuff={"text": "a generated haiku"}) - - await hello_world() - - assert "a generated haiku" in capsys.readouterr().out diff --git a/tests/unit/test_run_output.py b/tests/unit/test_run_output.py new file mode 100644 index 0000000..6313fe8 --- /dev/null +++ b/tests/unit/test_run_output.py @@ -0,0 +1,53 @@ +from mthds.runners.api.models import DictPipeOutputAbstract, DictRunResultExecute, DictStuffAbstract, DictWorkingMemoryAbstract +from pipelex_sdk.runs import RunResults + +from my_project.run_output import find_main_content, to_run_results + +ENTITIES_CONTENT = {"people": ["Marie Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} + + +class TestRunOutput: + def test_main_stuff_shape(self): + # Hosted durable runs: main_stuff carries the content directly. + results = RunResults(pipeline_run_id="run-1", main_stuff=ENTITIES_CONTENT) + assert find_main_content(results) == ENTITIES_CONTENT + + def test_pipe_output_shape(self): + # Bare-runner / blocking fallback: content sits inside the working memory root. + results = RunResults( + pipeline_run_id="run-2", + pipe_output={ + "working_memory": { + "root": {"extracted_entities": {"concept": "extract_entities.ExtractedEntities", "content": ENTITIES_CONTENT}}, + "aliases": {}, + }, + "pipeline_run_id": "run-2", + }, + ) + assert find_main_content(results) == ENTITIES_CONTENT + + def test_no_content_returns_none(self): + results = RunResults(pipeline_run_id="run-3") + assert find_main_content(results) is None + + def test_malformed_pipe_output_returns_none(self): + results = RunResults(pipeline_run_id="run-4", pipe_output={"working_memory": {"root": {"bad": "not-a-dict"}, "aliases": {}}}) + assert find_main_content(results) is None + + def test_to_run_results_normalizes_execute_response(self): + # The blocking `execute` response must land in the same RunResults shape + # the durable path returns, so find_main_content works on both. + execute_result = DictRunResultExecute( + pipeline_run_id="run-5", + pipe_output=DictPipeOutputAbstract( + working_memory=DictWorkingMemoryAbstract( + root={"extracted_entities": DictStuffAbstract(concept="extract_entities.ExtractedEntities", content=ENTITIES_CONTENT)}, + aliases={}, + ), + pipeline_run_id="run-5", + ), + ) + results = to_run_results(execute_result) + assert results.pipeline_run_id == "run-5" + assert results.main_stuff is None + assert find_main_content(results) == ENTITIES_CONTENT diff --git a/uv.lock b/uv.lock index ddeaf33..40b063d 100644 --- a/uv.lock +++ b/uv.lock @@ -7,6 +7,15 @@ resolution-markers = [ "python_full_version < '3.11'", ] +[[package]] +name = "annotated-doc" +version = "0.0.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/57/ba/046ceea27344560984e26a590f90bc7f4a75b06701f653222458922b558c/annotated_doc-0.0.4.tar.gz", hash = "sha256:fbcda96e87e9c92ad167c2e53839e57503ecfda18804ea28102353485033faa4", size = 7288, upload-time = "2025-11-10T22:07:42.062Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/1e/d3/26bf1008eb3d2daa8ef4cacc7f3bfdc11818d111f7e2d0201bc6e3b49d45/annotated_doc-0.0.4-py3-none-any.whl", hash = "sha256:571ac1dc6991c450b25a9c2d84a3705e2ae7a53467b5d111c24fa8baabbed320", size = 5303, upload-time = "2025-11-10T22:07:40.673Z" }, +] + [[package]] name = "annotated-types" version = "0.7.0" @@ -206,6 +215,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/3d/72cc9ec90bb80b5b1a65f0bb74a0f540195837baaf3b98c7fa4a7aa9718e/librt-0.6.3-cp314-cp314t-win_arm64.whl", hash = "sha256:afb39550205cc5e5c935762c6bf6a2bb34f7d21a68eadb25e2db7bf3593fecc0", size = 20246, upload-time = "2025-11-29T14:01:44.13Z" }, ] +[[package]] +name = "markdown-it-py" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mdurl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" }, +] + +[[package]] +name = "mdurl" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, +] + [[package]] name = "mthds" version = "0.7.1" @@ -229,8 +259,10 @@ name = "my-project" version = "0.10.0" source = { editable = "." } dependencies = [ + { name = "httpx" }, { name = "pipelex-sdk" }, { name = "python-dotenv" }, + { name = "typer" }, ] [package.optional-dependencies] @@ -240,21 +272,25 @@ dev = [ { name = "pyright" }, { name = "pytest" }, { name = "pytest-asyncio" }, + { name = "pytest-mock" }, { name = "pytest-sugar" }, { name = "ruff" }, ] [package.metadata] requires-dist = [ + { name = "httpx", specifier = ">=0.27.0" }, { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.11.2" }, { name = "pipelex-sdk", specifier = "==0.2.0" }, { name = "pipelex-tools", marker = "extra == 'dev'", specifier = ">=0.3.2" }, { name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.410" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=9.0.1" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.24.0" }, + { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.14.0" }, { name = "pytest-sugar", marker = "extra == 'dev'", specifier = ">=1.0.0" }, { name = "python-dotenv", specifier = ">=1.0.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.6.8" }, + { name = "typer", specifier = ">=0.15.0" }, ] provides-extras = ["dev"] @@ -567,6 +603,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" }, ] +[[package]] +name = "pytest-mock" +version = "3.15.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/68/14/eb014d26be205d38ad5ad20d9a80f7d201472e08167f0bb4361e251084a9/pytest_mock-3.15.1.tar.gz", hash = "sha256:1849a238f6f396da19762269de72cb1814ab44416fa73a8686deac10b0d87a0f", size = 34036, upload-time = "2025-09-16T16:37:27.081Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" }, +] + [[package]] name = "pytest-sugar" version = "1.1.1" @@ -589,6 +637,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0b/d7/1959b9648791274998a9c3526f6d0ec8fd2233e4d4acce81bbae76b44b2a/python_dotenv-1.2.2-py3-none-any.whl", hash = "sha256:1d8214789a24de455a8b8bd8ae6fe3c6b69a5e3d64aa8a8e5d68e694bbcb285a", size = 22101, upload-time = "2026-03-01T16:00:25.09Z" }, ] +[[package]] +name = "rich" +version = "15.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, +] + [[package]] name = "ruff" version = "0.14.7" @@ -624,6 +685,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6a/23/8146aad7d88f4fcb3a6218f41a60f6c2d4e3a72de72da1825dc7c8f7877c/semantic_version-2.10.0-py2.py3-none-any.whl", hash = "sha256:de78a3b8e0feda74cabc54aab2da702113e33ac9d9eb9d2389bcf1f58b7d9177", size = 15552, upload-time = "2022-05-26T13:35:21.206Z" }, ] +[[package]] +name = "shellingham" +version = "1.5.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/58/15/8b3609fd3830ef7b27b655beb4b4e9c62313a4e8da8c676e142cc210d58e/shellingham-1.5.4.tar.gz", hash = "sha256:8dbca0739d487e5bd35ab3ca4b36e11c4078f3a234bfce294b0a0291363404de", size = 10310, upload-time = "2023-10-24T04:13:40.426Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e0/f9/0595336914c5619e5f28a1fb793285925a8cd4b432c9da0a987836c7f822/shellingham-1.5.4-py2.py3-none-any.whl", hash = "sha256:7ecfff8f2fd72616f7481040475a65b2bf8af90a56c89140852d1120324e8686", size = 9755, upload-time = "2023-10-24T04:13:38.866Z" }, +] + [[package]] name = "termcolor" version = "3.2.0" @@ -691,6 +761,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/bd/75/8539d011f6be8e29f339c42e633aae3cb73bffa95dd0f9adec09b9c58e85/tomlkit-0.13.3-py3-none-any.whl", hash = "sha256:c89c649d79ee40629a9fda55f8ace8c6a1b42deb912b2a8fd8d942ddadb606b0", size = 38901, upload-time = "2025-06-05T07:13:43.546Z" }, ] +[[package]] +name = "typer" +version = "0.26.8" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "annotated-doc" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "rich" }, + { name = "shellingham" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7c/f7/68adc395201b20b872d68e975386832e8005ffeacedd43a1d837a32815be/typer-0.26.8.tar.gz", hash = "sha256:c244a6bd558886fe3f8780efb6bdd28bb9aff005a94eedebaa5cb32926fe2f7e", size = 202097, upload-time = "2026-06-26T09:22:45.705Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/80/87/b9fd69c92c6102a066e1b86a35243f53e70bd4c709f2a26d9f4fee4f4dc0/typer-0.26.8-py3-none-any.whl", hash = "sha256:3512ca79ac5c11113414b36e80281b872884477722440691c89d1112e321a49c", size = 122564, upload-time = "2026-06-26T09:22:44.72Z" }, +] + [[package]] name = "typing-extensions" version = "4.15.0" From cdfe328372ca84de7341c3fa0adfcdf532899744 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Sun, 5 Jul 2026 13:04:27 +0200 Subject: [PATCH 3/8] feat!: read run output via results.main_stuff; pin pipelex-sdk 0.3.0 Bumps pipelex-sdk to 0.3.0 (git-pinned to release/v0.3.0 while unreleased), which resolves the main output for you on both execution modes. Deletes the starter's find_main_content shape-guessing helper: to_run_results is now a trivial adapter and the CLI / narrower read results.main_stuff directly. A completed run with no main stuff raises the SDK's MissingMainStuffError. Bump 0.10.0 -> 0.11.0. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PibWhJ6mPWdrkMqZATZpkJ --- CHANGELOG.md | 3 ++ my_project/cli.py | 6 +-- my_project/examples/extract_entities.py | 12 ++--- my_project/run_output.py | 68 +++++------------------ pyproject.toml | 5 +- tests/unit/test_extract_entities.py | 21 +------- tests/unit/test_run_output.py | 72 ++++++++++--------------- uv.lock | 12 ++--- 8 files changed, 59 insertions(+), 140 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 380232a..ed1370c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ## [Unreleased] +## [v0.11.0] - 2026-07-05 + +- **Read a run's output with `results.main_stuff`.** Bumped to `pipelex-sdk` 0.3.0, which resolves the main output for you on both execution modes: `execute` returns a `PipelexExecuteResult` and the durable path a `RunResults`, and both expose a resolved `.main_stuff`. The starter's own output-extraction helper (`find_main_content`, which shape-guessed the working memory on the blocking path) is gone — `to_run_results` is now a trivial adapter, and the CLI / narrower read `results.main_stuff` directly. A completed run that delivers no main stuff raises the SDK's `MissingMainStuffError` instead of yielding `None`. - **Breaking:** renamed the env var `PIPELEX_API_URL` to `PIPELEX_BASE_URL` for consistency with the SDK's `base_url` naming. There is no read alias — update your `.env` / environment. ## [v0.10.0] - 2026-07-01 diff --git a/my_project/cli.py b/my_project/cli.py index e74b597..65efde9 100644 --- a/my_project/cli.py +++ b/my_project/cli.py @@ -19,7 +19,6 @@ from my_project.errors import present_error from my_project.examples import extract_entities as extract_entities_example -from my_project.run_output import find_main_content from my_project.runner import ( ExecutionMode, fetch_run_result, @@ -161,10 +160,7 @@ def _render_result_state(state: RunResultState) -> None: def _print_raw_results(results: RunResults) -> None: """Print the run's main content as JSON — generic, no per-example narrowing.""" - content: Any = find_main_content(results) - if content is None: - content = results.main_stuff if results.main_stuff is not None else results.pipe_output - output_console.print_json(data=content) + output_console.print_json(data=results.main_stuff) if __name__ == "__main__": diff --git a/my_project/examples/extract_entities.py b/my_project/examples/extract_entities.py index 5e9abe4..3dee53b 100644 --- a/my_project/examples/extract_entities.py +++ b/my_project/examples/extract_entities.py @@ -11,8 +11,6 @@ from pipelex_sdk.runs import RunResults from pydantic import BaseModel -from my_project.run_output import find_main_content - BUNDLE_PATH = Path(__file__).parent.parent / "methods" / "extract-entities" / "main.mthds" PIPE_CODE = "extract_entities" @@ -28,12 +26,10 @@ class ExtractedEntities(BaseModel): def parse(results: RunResults) -> ExtractedEntities: """Narrow a run result into a typed `ExtractedEntities`. + The SDK guarantees a resolved `results.main_stuff` for a completed run (it raises + `MissingMainStuffError` upstream otherwise), so this only validates the shape. + Raises: - RuntimeError: The run produced no output content at all. pydantic.ValidationError: The content doesn't match the concept's shape. """ - content = find_main_content(results) - if content is None: - msg = "The run returned no output content." - raise RuntimeError(msg) - return ExtractedEntities.model_validate(content) + return ExtractedEntities.model_validate(results.main_stuff) diff --git a/my_project/run_output.py b/my_project/run_output.py index f9e7a6d..8dd7bba 100644 --- a/my_project/run_output.py +++ b/my_project/run_output.py @@ -1,62 +1,22 @@ -"""Normalize the Pipelex API's run-output shapes into plain content dicts. +"""Give every execution mode the same result type for the CLI. -The API returns one of two opaque JSON shapes depending on the path taken: -hosted durable runs carry `main_stuff` (the main output's content directly), -while the blocking / bare-runner path carries `pipe_output` (a serialized -working memory). Everything downstream (the per-example `parse` functions) -goes through `find_main_content` so it never has to care which path ran. +The SDK hands back a `PipelexExecuteResult` from the blocking `execute` path and a +`RunResults` from the durable path — both already expose a resolved `.main_stuff` +(the main output's content, dug out of the working memory for you). `to_run_results` +just adapts the blocking result onto `RunResults` so every mode hands the CLI the same +type; from there, reading the output is simply `results.main_stuff`. """ -from typing import Any, cast - -from mthds.runners.api.models import DictRunResultExecute +from pipelex_sdk.execute_result import PipelexExecuteResult from pipelex_sdk.runs import RunResults -def find_main_content(results: RunResults) -> dict[str, Any] | None: - """Read the main output's content dict out of a run result. - - The Pipelex API returns one of two shapes (both opaque JSON), so we - normalize both here: - - Hosted runs carry `main_stuff` — the main output's content directly - (e.g. `{"people": [...], "orgs": [...], "dates": [...]}`). - - The bare-runner blocking fallback carries `pipe_output` - (`{"working_memory": {"root": {: {"content": ...}}}}`); we return - the first entry's content. - """ - main_stuff: Any = results.main_stuff - if isinstance(main_stuff, dict): - return cast("dict[str, Any]", main_stuff) - - pipe_output = results.pipe_output - if pipe_output is None: - return None - working_memory = pipe_output.get("working_memory") - if not isinstance(working_memory, dict): - return None - root = cast("dict[str, Any]", working_memory).get("root") - if not isinstance(root, dict): - return None - for entry in cast("dict[str, Any]", root).values(): - if not isinstance(entry, dict): - continue - content = cast("dict[str, Any]", entry).get("content") - if isinstance(content, dict): - return cast("dict[str, Any]", content) - return None - - -def to_run_results(result: DictRunResultExecute) -> RunResults: - """Normalize a blocking `execute` response into the lifecycle's `RunResults`. +def to_run_results(result: PipelexExecuteResult) -> RunResults: + """Adapt a blocking `execute` result onto the lifecycle's `RunResults`, so both + execution modes hand the CLI the same type. - The blocking path returns the runner's native shape (`pipe_output`); the - hosted-durable artifacts (`main_stuff`, `graph_spec`) don't exist on that - path and stay `None`. This mirrors what the SDK's `start_and_wait` does - internally, so both execution modes hand the same type to `find_main_content`. + `.main_stuff` is already resolved by the SDK on either path (it raises + `MissingMainStuffError` if a completed run named no main stuff), so there is no + working-memory digging left to do here. """ - return RunResults( - pipeline_run_id=result.pipeline_run_id, - main_stuff=None, - graph_spec=None, - pipe_output=result.pipe_output.model_dump(), - ) + return RunResults(pipeline_run_id=result.pipeline_run_id, main_stuff=result.main_stuff) diff --git a/pyproject.toml b/pyproject.toml index 7e6d813..1709da9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "my-project" -version = "0.10.0" +version = "0.11.0" description = "Replace this with your project description" # authors = [{ name = "Your Name", email = "your.email@example.com" }] license = "MIT" @@ -18,7 +18,8 @@ classifiers = [ dependencies = [ "httpx>=0.27.0", - "pipelex-sdk==0.2.0", + # Pinned to the unreleased SDK branch by git while we iterate before publishing pipelex-sdk 0.3.0. + "pipelex-sdk @ git+ssh://git@github.com/Pipelex/pipelex-sdk-python.git@release/v0.3.0", "python-dotenv>=1.0.0", "typer>=0.15.0", ] diff --git a/tests/unit/test_extract_entities.py b/tests/unit/test_extract_entities.py index e6d9705..5a87c7c 100644 --- a/tests/unit/test_extract_entities.py +++ b/tests/unit/test_extract_entities.py @@ -8,33 +8,14 @@ class TestExtractEntitiesParse: - def test_parse_main_stuff_shape(self): + def test_parse_main_stuff(self): results = RunResults(pipeline_run_id="run-1", main_stuff=ENTITIES_CONTENT) entities = parse(results) assert entities.people == ["Marie Curie", "Pierre Curie"] assert entities.orgs == ["University of Paris"] assert entities.dates == ["1906"] - def test_parse_pipe_output_shape(self): - results = RunResults( - pipeline_run_id="run-2", - pipe_output={ - "working_memory": { - "root": {"extracted_entities": {"concept": "extract_entities.ExtractedEntities", "content": ENTITIES_CONTENT}}, - "aliases": {}, - }, - "pipeline_run_id": "run-2", - }, - ) - entities = parse(results) - assert entities.people == ["Marie Curie", "Pierre Curie"] - def test_parse_shape_mismatch_raises(self): results = RunResults(pipeline_run_id="run-3", main_stuff={"people": "not-a-list", "orgs": [], "dates": []}) with pytest.raises(ValidationError): parse(results) - - def test_parse_no_content_raises(self): - results = RunResults(pipeline_run_id="run-4") - with pytest.raises(RuntimeError, match="no output content"): - parse(results) diff --git a/tests/unit/test_run_output.py b/tests/unit/test_run_output.py index 6313fe8..fd8598d 100644 --- a/tests/unit/test_run_output.py +++ b/tests/unit/test_run_output.py @@ -1,53 +1,39 @@ -from mthds.runners.api.models import DictPipeOutputAbstract, DictRunResultExecute, DictStuffAbstract, DictWorkingMemoryAbstract -from pipelex_sdk.runs import RunResults +import pytest +from pipelex_sdk.errors import MissingMainStuffError +from pipelex_sdk.execute_result import PipelexExecuteResult -from my_project.run_output import find_main_content, to_run_results +from my_project.run_output import to_run_results ENTITIES_CONTENT = {"people": ["Marie Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} -class TestRunOutput: - def test_main_stuff_shape(self): - # Hosted durable runs: main_stuff carries the content directly. - results = RunResults(pipeline_run_id="run-1", main_stuff=ENTITIES_CONTENT) - assert find_main_content(results) == ENTITIES_CONTENT - - def test_pipe_output_shape(self): - # Bare-runner / blocking fallback: content sits inside the working memory root. - results = RunResults( - pipeline_run_id="run-2", - pipe_output={ - "working_memory": { - "root": {"extracted_entities": {"concept": "extract_entities.ExtractedEntities", "content": ENTITIES_CONTENT}}, - "aliases": {}, - }, - "pipeline_run_id": "run-2", - }, - ) - assert find_main_content(results) == ENTITIES_CONTENT - - def test_no_content_returns_none(self): - results = RunResults(pipeline_run_id="run-3") - assert find_main_content(results) is None +def _execute_result(*, main_stuff_name: str, root: dict[str, object]) -> PipelexExecuteResult: + return PipelexExecuteResult.model_validate( + { + "pipeline_run_id": "run-5", + "main_stuff_name": main_stuff_name, + "pipe_output": {"working_memory": {"root": root, "aliases": {}}, "pipeline_run_id": "run-5"}, + } + ) - def test_malformed_pipe_output_returns_none(self): - results = RunResults(pipeline_run_id="run-4", pipe_output={"working_memory": {"root": {"bad": "not-a-dict"}, "aliases": {}}}) - assert find_main_content(results) is None - def test_to_run_results_normalizes_execute_response(self): - # The blocking `execute` response must land in the same RunResults shape - # the durable path returns, so find_main_content works on both. - execute_result = DictRunResultExecute( - pipeline_run_id="run-5", - pipe_output=DictPipeOutputAbstract( - working_memory=DictWorkingMemoryAbstract( - root={"extracted_entities": DictStuffAbstract(concept="extract_entities.ExtractedEntities", content=ENTITIES_CONTENT)}, - aliases={}, - ), - pipeline_run_id="run-5", - ), +class TestRunOutput: + def test_to_run_results_surfaces_resolved_main_stuff(self): + # The blocking `execute` result resolves `.main_stuff` out of its working memory; + # `to_run_results` lands it on the same `RunResults` shape the durable path returns. + execute_result = _execute_result( + main_stuff_name="extracted_entities", + root={"extracted_entities": {"concept": "extract_entities.ExtractedEntities", "content": ENTITIES_CONTENT}}, ) results = to_run_results(execute_result) assert results.pipeline_run_id == "run-5" - assert results.main_stuff is None - assert find_main_content(results) == ENTITIES_CONTENT + assert results.main_stuff == ENTITIES_CONTENT + + def test_to_run_results_raises_when_main_stuff_unlocatable(self): + # `main_stuff_name` names a stuff absent from the working-memory root — a hard fail. + execute_result = _execute_result( + main_stuff_name="missing", + root={"other": {"concept": "native.Text", "content": {}}}, + ) + with pytest.raises(MissingMainStuffError): + to_run_results(execute_result) diff --git a/uv.lock b/uv.lock index 40b063d..31daba4 100644 --- a/uv.lock +++ b/uv.lock @@ -256,7 +256,7 @@ wheels = [ [[package]] name = "my-project" -version = "0.10.0" +version = "0.11.0" source = { editable = "." } dependencies = [ { name = "httpx" }, @@ -281,7 +281,7 @@ dev = [ requires-dist = [ { name = "httpx", specifier = ">=0.27.0" }, { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.11.2" }, - { name = "pipelex-sdk", specifier = "==0.2.0" }, + { name = "pipelex-sdk", git = "ssh://git@github.com/Pipelex/pipelex-sdk-python.git?rev=release%2Fv0.3.0" }, { name = "pipelex-tools", marker = "extra == 'dev'", specifier = ">=0.3.2" }, { name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.410" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=9.0.1" }, @@ -378,8 +378,8 @@ wheels = [ [[package]] name = "pipelex-sdk" -version = "0.2.0" -source = { registry = "https://pypi.org/simple" } +version = "0.3.0" +source = { git = "ssh://git@github.com/Pipelex/pipelex-sdk-python.git?rev=release%2Fv0.3.0#68e2b8efe03b3110bf72bd62b8d44e6a588bc229" } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" }, { name = "httpx" }, @@ -387,10 +387,6 @@ dependencies = [ { name = "pydantic" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/53/48/585dd15cb559fe3a83b39eb5279ecca99446f388777a1d97d1823606c0d3/pipelex_sdk-0.2.0.tar.gz", hash = "sha256:2df844f6e79c5e84465855452b0838f29db1d4356897fc6b08b057c837364539", size = 116730, upload-time = "2026-07-02T22:12:07.891Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/dd/31/304cd4cbcb70499563a3dd34f1120bad6c337f314c45df674b5caeb611f7/pipelex_sdk-0.2.0-py3-none-any.whl", hash = "sha256:ea94d40cfcf56fe234294d6324c4ac626c1658c61f3a15d698b968ba9fee2e0d", size = 31459, upload-time = "2026-07-02T22:12:06.689Z" }, -] [[package]] name = "pipelex-tools" From a446046d207dfb06f6637a217f2dc401e5f01f8d Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Sun, 5 Jul 2026 13:26:40 +0200 Subject: [PATCH 4/8] refactor: drop the run_output adapter module; adapt blocking result inline Now that the SDK resolves .main_stuff on both the blocking (PipelexExecuteResult) and durable (RunResults) result types, my_project/run_output.py earned nothing: its find_main_content was already deleted and to_run_results was a one-line repackage. Inlined that adaptation into runner.run_blocking and removed the module + its test. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PibWhJ6mPWdrkMqZATZpkJ --- CHANGELOG.md | 2 +- my_project/run_output.py | 22 -------------------- my_project/runner.py | 7 ++++--- tests/unit/test_run_output.py | 39 ----------------------------------- 4 files changed, 5 insertions(+), 65 deletions(-) delete mode 100644 my_project/run_output.py delete mode 100644 tests/unit/test_run_output.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ed1370c..a197cee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ ## [v0.11.0] - 2026-07-05 -- **Read a run's output with `results.main_stuff`.** Bumped to `pipelex-sdk` 0.3.0, which resolves the main output for you on both execution modes: `execute` returns a `PipelexExecuteResult` and the durable path a `RunResults`, and both expose a resolved `.main_stuff`. The starter's own output-extraction helper (`find_main_content`, which shape-guessed the working memory on the blocking path) is gone — `to_run_results` is now a trivial adapter, and the CLI / narrower read `results.main_stuff` directly. A completed run that delivers no main stuff raises the SDK's `MissingMainStuffError` instead of yielding `None`. +- **Read a run's output with `results.main_stuff`.** Bumped to `pipelex-sdk` 0.3.0, which resolves the main output for you on both execution modes: `execute` returns a `PipelexExecuteResult` and the durable path a `RunResults`, and both expose a resolved `.main_stuff`. The starter's whole output-extraction module (`my_project/run_output.py` — `find_main_content` shape-guessing + the `to_run_results` adapter) is gone; the CLI and the narrower read `results.main_stuff` directly, and the blocking `execute` result is adapted onto `RunResults` inline in the runner. A completed run that delivers no main stuff raises the SDK's `MissingMainStuffError` instead of yielding `None`. - **Breaking:** renamed the env var `PIPELEX_API_URL` to `PIPELEX_BASE_URL` for consistency with the SDK's `base_url` naming. There is no read alias — update your `.env` / environment. ## [v0.10.0] - 2026-07-01 diff --git a/my_project/run_output.py b/my_project/run_output.py deleted file mode 100644 index 8dd7bba..0000000 --- a/my_project/run_output.py +++ /dev/null @@ -1,22 +0,0 @@ -"""Give every execution mode the same result type for the CLI. - -The SDK hands back a `PipelexExecuteResult` from the blocking `execute` path and a -`RunResults` from the durable path — both already expose a resolved `.main_stuff` -(the main output's content, dug out of the working memory for you). `to_run_results` -just adapts the blocking result onto `RunResults` so every mode hands the CLI the same -type; from there, reading the output is simply `results.main_stuff`. -""" - -from pipelex_sdk.execute_result import PipelexExecuteResult -from pipelex_sdk.runs import RunResults - - -def to_run_results(result: PipelexExecuteResult) -> RunResults: - """Adapt a blocking `execute` result onto the lifecycle's `RunResults`, so both - execution modes hand the CLI the same type. - - `.main_stuff` is already resolved by the SDK on either path (it raises - `MissingMainStuffError` if a completed run named no main stuff), so there is no - working-memory digging left to do here. - """ - return RunResults(pipeline_run_id=result.pipeline_run_id, main_stuff=result.main_stuff) diff --git a/my_project/runner.py b/my_project/runner.py index 4e69bd0..66022b3 100644 --- a/my_project/runner.py +++ b/my_project/runner.py @@ -21,8 +21,6 @@ from pipelex_sdk.runs import PollInfo, RunRead, RunResults, RunResultState, WaitForResultOptions from rich.console import Console -from my_project.run_output import to_run_results - # Progress and lifecycle chatter go to stderr so stdout stays pipeable. progress_console = Console(stderr=True) @@ -43,7 +41,10 @@ async def run_blocking(*, pipe_code: str, bundle: str, inputs: dict[str, Any] | async with PipelexAPIClient() as client: with progress_console.status("Running (blocking)…"): result = await client.execute(pipe_code=pipe_code, mthds_contents=[bundle], inputs=inputs) - return to_run_results(result) + # Adapt the blocking `execute` result onto `RunResults` so both modes return one type; + # `.main_stuff` is already resolved by the SDK (it raises `MissingMainStuffError` if a + # completed run named no main stuff). + return RunResults(pipeline_run_id=result.pipeline_run_id, main_stuff=result.main_stuff) async def run_durable_attended(*, pipe_code: str, bundle: str, inputs: dict[str, Any] | None = None) -> RunResults: diff --git a/tests/unit/test_run_output.py b/tests/unit/test_run_output.py deleted file mode 100644 index fd8598d..0000000 --- a/tests/unit/test_run_output.py +++ /dev/null @@ -1,39 +0,0 @@ -import pytest -from pipelex_sdk.errors import MissingMainStuffError -from pipelex_sdk.execute_result import PipelexExecuteResult - -from my_project.run_output import to_run_results - -ENTITIES_CONTENT = {"people": ["Marie Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} - - -def _execute_result(*, main_stuff_name: str, root: dict[str, object]) -> PipelexExecuteResult: - return PipelexExecuteResult.model_validate( - { - "pipeline_run_id": "run-5", - "main_stuff_name": main_stuff_name, - "pipe_output": {"working_memory": {"root": root, "aliases": {}}, "pipeline_run_id": "run-5"}, - } - ) - - -class TestRunOutput: - def test_to_run_results_surfaces_resolved_main_stuff(self): - # The blocking `execute` result resolves `.main_stuff` out of its working memory; - # `to_run_results` lands it on the same `RunResults` shape the durable path returns. - execute_result = _execute_result( - main_stuff_name="extracted_entities", - root={"extracted_entities": {"concept": "extract_entities.ExtractedEntities", "content": ENTITIES_CONTENT}}, - ) - results = to_run_results(execute_result) - assert results.pipeline_run_id == "run-5" - assert results.main_stuff == ENTITIES_CONTENT - - def test_to_run_results_raises_when_main_stuff_unlocatable(self): - # `main_stuff_name` names a stuff absent from the working-memory root — a hard fail. - execute_result = _execute_result( - main_stuff_name="missing", - root={"other": {"concept": "native.Text", "content": {}}}, - ) - with pytest.raises(MissingMainStuffError): - to_run_results(execute_result) From 43fd0295d058f0515f104e5d2c7e607cc37f8eab Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Sun, 5 Jul 2026 13:36:17 +0200 Subject: [PATCH 5/8] chore: re-lock pipelex-sdk to release/v0.3.0 tip (458dcc7) Picks up the SDK's main_stuff_name typed-field cleanup (behavior-neutral). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PibWhJ6mPWdrkMqZATZpkJ --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index 31daba4..a72d981 100644 --- a/uv.lock +++ b/uv.lock @@ -379,7 +379,7 @@ wheels = [ [[package]] name = "pipelex-sdk" version = "0.3.0" -source = { git = "ssh://git@github.com/Pipelex/pipelex-sdk-python.git?rev=release%2Fv0.3.0#68e2b8efe03b3110bf72bd62b8d44e6a588bc229" } +source = { git = "ssh://git@github.com/Pipelex/pipelex-sdk-python.git?rev=release%2Fv0.3.0#458dcc79c56b146493d9d1c8822317583df0a6d3" } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" }, { name = "httpx" }, From c1d26596bc72e057f9a023a00355c8d7947a3762 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Sun, 5 Jul 2026 15:51:17 +0200 Subject: [PATCH 6/8] docs: rewrite README/CLAUDE.md around the my-project CLI; drop internal planning docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The starter was rewritten from a single hello_world.py example to an extract-entities Typer CLI, but the user-facing docs still described the deleted module. Rewrite the README and CLAUDE.md around what actually exists: the my-project console script, the extract-entities command, the durable/blocking execution modes, and the runs status|result|wait lifecycle. The quick start now runs `uv run my-project extract-entities ""` — the old first command (`python -m my_project.hello_world`) errored out. Drop the stale references to start_and_wait usage and the removed find_main_content normalizer. Also remove the internal-only planning docs (TODOS.md, wip/) that must not ship in a "Use this template" repo, mirroring the JS starter's cleanup. CHANGELOG: fold both fixes into the unreleased v0.11.0 section. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01PibWhJ6mPWdrkMqZATZpkJ --- CHANGELOG.md | 2 ++ CLAUDE.md | 8 ++++---- README.md | 43 ++++++++++++++++++++++++++++--------------- 3 files changed, 34 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a197cee..701b81b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ - **Read a run's output with `results.main_stuff`.** Bumped to `pipelex-sdk` 0.3.0, which resolves the main output for you on both execution modes: `execute` returns a `PipelexExecuteResult` and the durable path a `RunResults`, and both expose a resolved `.main_stuff`. The starter's whole output-extraction module (`my_project/run_output.py` — `find_main_content` shape-guessing + the `to_run_results` adapter) is gone; the CLI and the narrower read `results.main_stuff` directly, and the blocking `execute` result is adapted onto `RunResults` inline in the runner. A completed run that delivers no main stuff raises the SDK's `MissingMainStuffError` instead of yielding `None`. - **Breaking:** renamed the env var `PIPELEX_API_URL` to `PIPELEX_BASE_URL` for consistency with the SDK's `base_url` naming. There is no read alias — update your `.env` / environment. +- **Fixed:** rewrote the README and `CLAUDE.md` around the actual `my-project` CLI (the `extract-entities` command, the durable/blocking execution modes, and the `runs status|result|wait` lifecycle). They still described the removed `hello_world` module, `start_and_wait` usage, and the `find_main_content` normalizer, so the quick start's first command errored out for a fresh user. +- **Repository:** removed internal-only planning docs (`TODOS.md`, `wip/`) that must not ship in a "Use this template" repo. ## [v0.10.0] - 2026-07-01 diff --git a/CLAUDE.md b/CLAUDE.md index faf1a4d..bfd4113 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -35,16 +35,16 @@ Run specific tests (local only): `make tp TEST=test_function_name` This starter calls the **hosted Pipelex API** via the `pipelex-sdk` package (`PipelexAPIClient`) — it does **not** run Pipelex as a local library. The `.mthds` bundle is read from disk and sent to the API as content (`mthds_contents`); the API runs the method and returns the output. - Credentials/endpoint come from `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` (see `.env.example`). `python-dotenv` loads `.env` when running the CLI or tests. -- `my_project/hello_world.py` uses `client.start_and_wait(...)` — the durable start-and-poll path (survives the hosted gateway's ~30s cap, self-heals to blocking `execute` on a bare runner). -- Output is loosely-typed JSON: hosted runs carry `main_stuff`; the bare-runner fallback carries `pipe_output`. `find_main_content()` normalizes both. +- The `my-project` CLI (`my_project/cli.py`) is a Typer app; `my_project/runner.py` dispatches each run by execution mode — `blocking` (`client.execute`), durable attended (`client.start` + `client.wait_for_result`), and durable detached (`client.start` only, resumed via `my-project runs status|result|wait `). It branches on mode explicitly rather than using the SDK's `start_and_wait` self-healing one-liner, because teaching the mode difference is the point. +- The SDK resolves the main output on both modes: `client.execute` returns a `PipelexExecuteResult` and the durable path a `RunResults`, both exposing a resolved `.main_stuff` (a completed run with no main stuff raises `MissingMainStuffError`). Per-example narrowing lives in `my_project/examples/` — `extract_entities.parse()` validates `results.main_stuff` into a typed `ExtractedEntities` model. SDK errors are mapped to CLI-facing messages + hints in `my_project/errors.py`. ## Project Structure - Package: `my_project/` (Python 3.10+, target 3.11) -- Tests: `tests/` (integration = offline boot/bundle checks + API `validate`; e2e = full run via the API) +- Tests: `tests/` (unit = offline CLI/example/error-mapping tests; integration = offline boot/bundle checks + API `validate`; e2e = full run via the API) - Dependency manager: uv (>=0.7.2) - Pipelex dependency: `pipelex-sdk` package from PyPI (the API client — see pyproject.toml). The `pipelex` runtime is **not** a dependency. -- `.mthds` files: Pipelex method definition files in `my_project/` +- `.mthds` files: Pipelex method definition files in `my_project/methods//main.mthds` ## Test markers diff --git a/README.md b/README.md index fd3adeb..f01516f 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ A minimal Python CLI starter that calls the [Pipelex](https://pipelex.com) API via the [`pipelex-sdk`](https://pypi.org/project/pipelex-sdk/) SDK to run AI methods (`.mthds` bundles) — no local Pipelex runtime required. -It ships one demo pipeline, `my_project/hello_world.mthds`, which asks an LLM to write a haiku about "Hello World" and prints it. +It ships one demo method, `extract-entities` (`my_project/methods/extract-entities/main.mthds`), exposed through a `my-project` CLI. Given a piece of text it asks an LLM to pull out the people, organizations, and dates it mentions, and prints them as JSON. ### Use this template @@ -31,38 +31,51 @@ cp .env.example .env # edit .env and set PIPELEX_API_KEY (and PIPELEX_BASE_URL if self-hosting) make install # create the venv and install deps with uv -python -m my_project.hello_world # run the hello_world example against the API +uv run my-project extract-entities "Alice from Acme met Bob on May 3rd, 2026." ``` -That prints the generated haiku. +That prints the extracted people, organizations, and dates as JSON. (`uv run` finds the project's venv; activate it with `source .venv/bin/activate` if you'd rather drop the prefix and just call `my-project ...`.) ## Project structure ``` my_project/ - hello_world.mthds # the method bundle: text → { text } haiku - hello_world.py # the CLI entry point that runs the bundle via the SDK + cli.py # the `my-project` Typer CLI (console-script entry point) + runner.py # execution-mode dispatch: blocking / durable attended / detached + errors.py # maps SDK errors to CLI messages + hints + examples/ + extract_entities.py # the "copy me" unit: bundle path, output model, parse() narrower + methods/ + extract-entities/main.mthds # the method bundle: text → { people, orgs, dates } tests/ - integration/ # offline boot/bundle checks + API validate (pipelex_api) - e2e/ # full run against the API (inference) -.env.example # PIPELEX_BASE_URL + PIPELEX_API_KEY + unit/ # offline CLI / example / error-mapping tests + integration/ # offline boot/bundle checks + API validate (pipelex_api) + e2e/ # full run against the API (inference) +.env.example # PIPELEX_BASE_URL + PIPELEX_API_KEY ``` ## How it works -`my_project/hello_world.py`: +`my-project extract-entities ""`: -1. Reads the `.mthds` bundle from disk (`BUNDLE_PATH`). -2. Constructs a `PipelexAPIClient`, which reads `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` from the environment. -3. Calls `start_and_wait(pipe_code="hello_world", mthds_contents=[bundle])` — the durable start-and-poll path that survives the hosted gateway's ~30s synchronous cap and self-heals to a blocking `execute` against a bare self-hosted runner. -4. Reads the main output's content (`{"text": ...}`) out of the loosely-typed result and prints it. +1. Reads the `.mthds` bundle from disk (`extract_entities.BUNDLE_PATH`) and constructs a `PipelexAPIClient`, which picks up `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` from the environment. +2. Runs the pipe against the API in one of **two execution modes** (`--mode`, env var `PIPELEX_EXECUTION_MODE`): + - **durable** (default) — `client.start()` then poll the run to completion (`client.wait_for_result`). Survives the hosted gateway's ~30s synchronous cap, so long runs succeed; the run id is printed first, so a Ctrl-C leaves it executing server-side and you can resume with `my-project runs wait `. + - **blocking** — a single `client.execute()` call. Simpler, but behind the hosted gateway a run over ~30s is cut off and surfaces a clear timeout error pointing you at durable mode. +3. Reads the resolved main output — the SDK exposes `results.main_stuff` on both modes — and the example's `parse()` narrower validates it into a typed `ExtractedEntities` model, printed as JSON. -The `.mthds` bundle is sent to the API as content (`mthds_contents`), so nothing about the method needs to live in the runtime — edit `hello_world.mthds` and re-run. +The `.mthds` bundle is sent to the API as content (`mthds_contents`), so nothing about the method needs to live in the runtime — edit `methods/extract-entities/main.mthds` and re-run. + +`my_project/runner.py` deliberately branches on the mode explicitly rather than calling the SDK's `start_and_wait()` self-healing one-liner (the production shortcut when you don't care which mode) — teaching the difference between the two paths is the point of this starter. Pass `--detach` (durable only) to start a run and return immediately, then pick it back up later with `my-project runs status|result|wait `. ## Useful commands ```bash -python -m my_project.hello_world # run the hello_world example +uv run my-project extract-entities "Alice from Acme met Bob on May 3rd, 2026." # run the demo method +uv run my-project extract-entities --file notes.txt # read the input text from a file +uv run my-project extract-entities "…" --mode blocking # single synchronous call (~30s cap on hosted) +uv run my-project extract-entities "…" --detach # start a durable run, print its id, return +uv run my-project runs wait # resume a detached run to completion (also: runs status / runs result) make validate # lint/validate the .mthds bundle with plxt (offline) make agent-check # fix-imports + format + lint + pyright + mypy make agent-test # run the offline test suite (silent on success) From 59dcb2bd4bb0cc42dedabba4a1c300d21ff5508d Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Sun, 5 Jul 2026 17:23:21 +0200 Subject: [PATCH 7/8] chore: swap pipelex-sdk pin from git ref to the published PyPI 0.3.0 pipelex-sdk 0.3.0 published to PyPI now that pipelex-sdk-python#6 merged and the repo is public. The git+ssh pin couldn't be cloned by CI runners (no SSH key), which is exactly the failure this resolves. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01PibWhJ6mPWdrkMqZATZpkJ --- pyproject.toml | 3 +-- uv.lock | 8 ++++++-- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 1709da9..3bcaf3b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,8 +18,7 @@ classifiers = [ dependencies = [ "httpx>=0.27.0", - # Pinned to the unreleased SDK branch by git while we iterate before publishing pipelex-sdk 0.3.0. - "pipelex-sdk @ git+ssh://git@github.com/Pipelex/pipelex-sdk-python.git@release/v0.3.0", + "pipelex-sdk>=0.3.0", "python-dotenv>=1.0.0", "typer>=0.15.0", ] diff --git a/uv.lock b/uv.lock index a72d981..92d98e1 100644 --- a/uv.lock +++ b/uv.lock @@ -281,7 +281,7 @@ dev = [ requires-dist = [ { name = "httpx", specifier = ">=0.27.0" }, { name = "mypy", marker = "extra == 'dev'", specifier = ">=1.11.2" }, - { name = "pipelex-sdk", git = "ssh://git@github.com/Pipelex/pipelex-sdk-python.git?rev=release%2Fv0.3.0" }, + { name = "pipelex-sdk", specifier = ">=0.3.0" }, { name = "pipelex-tools", marker = "extra == 'dev'", specifier = ">=0.3.2" }, { name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.410" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=9.0.1" }, @@ -379,7 +379,7 @@ wheels = [ [[package]] name = "pipelex-sdk" version = "0.3.0" -source = { git = "ssh://git@github.com/Pipelex/pipelex-sdk-python.git?rev=release%2Fv0.3.0#458dcc79c56b146493d9d1c8822317583df0a6d3" } +source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" }, { name = "httpx" }, @@ -387,6 +387,10 @@ dependencies = [ { name = "pydantic" }, { name = "typing-extensions" }, ] +sdist = { url = "https://files.pythonhosted.org/packages/d8/22/56770109ac02047147830aef78972550aa2939132f964879cd29def56965/pipelex_sdk-0.3.0.tar.gz", hash = "sha256:02f9deb692125fc5843d50cb9b6f698c331e5335868ad0b4f1e4eb915933ce75", size = 120328, upload-time = "2026-07-05T14:28:35.853Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9d/01/bb84efcb45df99ff0a27e582e51dcface11de1c641fa1be524151aec8d00/pipelex_sdk-0.3.0-py3-none-any.whl", hash = "sha256:a8af6b8298b6f509fec1dca41830c121e2605c6e55cc6c6ba5d2cdc546e99c7d", size = 33602, upload-time = "2026-07-05T14:28:34.524Z" }, +] [[package]] name = "pipelex-tools" From 325511a40fbec28bf6836ed9d99b03080364665c Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Sun, 5 Jul 2026 18:25:31 +0200 Subject: [PATCH 8/8] fix: type extract-entities list fields as item_type = "text" The demo bundle declared people/orgs/dates as bare type = "list" with no item_type, so the runtime built them as List[Any] instead of the list[str] the ExtractedEntities model mirrors. Add item_type = "text" to all three. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_014mFVgTkeUHpKPBqAbArWLv --- CHANGELOG.md | 1 + my_project/methods/extract-entities/main.mthds | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 701b81b..4793bf7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ ## [v0.11.0] - 2026-07-05 +- **Fixed:** the demo bundle's list fields (`people`/`orgs`/`dates`) now declare `item_type = "text"` so the output is typed as `list[str]`, matching the `ExtractedEntities` model. Without it the runtime built the fields as `List[Any]`. - **Read a run's output with `results.main_stuff`.** Bumped to `pipelex-sdk` 0.3.0, which resolves the main output for you on both execution modes: `execute` returns a `PipelexExecuteResult` and the durable path a `RunResults`, and both expose a resolved `.main_stuff`. The starter's whole output-extraction module (`my_project/run_output.py` — `find_main_content` shape-guessing + the `to_run_results` adapter) is gone; the CLI and the narrower read `results.main_stuff` directly, and the blocking `execute` result is adapted onto `RunResults` inline in the runner. A completed run that delivers no main stuff raises the SDK's `MissingMainStuffError` instead of yielding `None`. - **Breaking:** renamed the env var `PIPELEX_API_URL` to `PIPELEX_BASE_URL` for consistency with the SDK's `base_url` naming. There is no read alias — update your `.env` / environment. - **Fixed:** rewrote the README and `CLAUDE.md` around the actual `my-project` CLI (the `extract-entities` command, the durable/blocking execution modes, and the `runs status|result|wait` lifecycle). They still described the removed `hello_world` module, `start_and_wait` usage, and the `find_main_content` normalizer, so the quick start's first command errored out for a fresh user. diff --git a/my_project/methods/extract-entities/main.mthds b/my_project/methods/extract-entities/main.mthds index ebcbbc6..ec2c958 100644 --- a/my_project/methods/extract-entities/main.mthds +++ b/my_project/methods/extract-entities/main.mthds @@ -6,9 +6,9 @@ main_pipe = "extract_entities" description = "Named entities extracted from a piece of text." [concept.ExtractedEntities.structure] -people = { type = "list", description = "Names of people mentioned in the text", required = true } -orgs = { type = "list", description = "Names of organizations mentioned in the text", required = true } -dates = { type = "list", description = "Dates or time references mentioned in the text", required = true } +people = { type = "list", item_type = "text", description = "Names of people mentioned in the text", required = true } +orgs = { type = "list", item_type = "text", description = "Names of organizations mentioned in the text", required = true } +dates = { type = "list", item_type = "text", description = "Dates or time references mentioned in the text", required = true } [pipe.extract_entities] type = "PipeLLM"