feat: local prompt pulling

README.md

@@ -1,4 +1,4 @@
-# langfuse-cli
+# bto-langfuse-cli
 
 A set of CLI tools to interact with Langfuse, especially for prompt management.
 
@@ -16,7 +16,7 @@ Or you can install it using pip if you have the package:
 pip install .
 ```
 
-The CLI command is named `langfuse-cli`.
+The CLI command is named `bto-langfuse-cli`.
 
 ## Configuration
 
@@ -46,7 +46,7 @@ The `promote` command allows you to copy a label from one version of all prompts
 
 **Command:**
 ```bash
-langfuse-cli prompt promote <src_label> <target_label> [--apply]
+bto-langfuse-cli prompt promote <src_label> <target_label> [--apply]
 ```
 
 **Options:**
@@ -56,20 +56,52 @@ langfuse-cli prompt promote <src_label> <target_label> [--apply]
 
 1. **Dry run / Confirmation mode:**
    ```bash
-   langfuse-cli prompt promote dev uat
+   bto-langfuse-cli prompt promote dev uat
    ```
    This will show a plan of which prompts will be updated (adding the `uat` label to all prompts that currently have the `dev` label) and ask for confirmation before applying.
 
 2. **Auto-apply mode:**
    ```bash
-   langfuse-cli prompt promote dev uat --apply
+   bto-langfuse-cli prompt promote dev uat --apply
    ```
    This will automatically update the labels for all prompts matching the source label.
 
+#### Pull Prompts
+
+The `pull` command downloads prompts from Langfuse to your local filesystem based on a specific label.
+
+**Command:**
+```bash
+bto-langfuse-cli prompt pull <label> [type] [OPTIONS]
+```
+
+**Arguments:**
+- `<label>`: The label to pull prompts from (e.g., `production`, `dev`). **(Required)**
+- `[type]`: The type of prompt to pull. Can be `text` or `chat`. If not specified, both types are pulled.
+
+**Options:**
+- `-o, --output-dir <DIRECTORY>`: Directory to save the prompt files to. Defaults to `data/prompts`.
+
+**Output Format (`md`):**
+- **Text Prompts**: Saved as `.md` files containing a YAML frontmatter (with metadata like name, version, tags, labels, and last updated date) followed by the raw text prompt.
+- **Chat Prompts**: Saved as `.md` files containing the same YAML frontmatter, followed by the chat messages serialized using XML-like tags (e.g., `<system>`, `<user>`).
+
+**Examples:**
+
+1. **Pull all prompts with a specific label:**
+   ```bash
+   bto-langfuse-cli prompt pull production
+   ```
+
+2. **Pull only chat prompts to a custom directory:**
+   ```bash
+   bto-langfuse-cli prompt pull dev chat -o ./my-local-prompts
+   ```
+
 ## Development
 
 To run the CLI during development:
 
 ```bash
-uv run langfuse-cli --help
+uv run bto-langfuse-cli --help
 ```

src/bto_langfuse_cli/commands/prompt.py

@@ -2,23 +2,25 @@
 
 from __future__ import annotations
 
-from typing import Annotated
+from pathlib import Path
+from typing import Annotated, Literal
 
 import typer
 
 from bto_langfuse_cli.langfuse.langfuse_service import LangfuseService
 from bto_langfuse_cli.langfuse.print_plan import print_plan
+from bto_langfuse_cli.langfuse.prompt_exporter import MarkdownPromptExporter
 
 prompt_app = typer.Typer(help="Manage Langfuse prompts.")
 
 
 @prompt_app.command("promote")
 def prompt_promote(
-        args: Annotated[
-            tuple[str, str],
-            typer.Argument(help="src_label target_label)."),
-        ],
-        apply: Annotated[bool, typer.Option(help="Apply the promotion automatically")] = False,
+    args: Annotated[
+        tuple[str, str],
+        typer.Argument(help="src_label target_label)."),
+    ],
+    apply: Annotated[bool, typer.Option(help="Apply the promotion automatically")] = False,
 ) -> None:
     """Promote a label to another label.
 
@@ -33,3 +35,56 @@ def prompt_promote(
     if plan.has_changes():
         if apply or typer.confirm(f"Apply plan ?"):
             langfuse.apply(plan, to_label)
+
+
+@prompt_app.command("pull")
+def prompt_pull(
+    label: Annotated[
+        str,
+        typer.Argument(help="The label to pull prompt from."),
+    ],
+    output_dir: Annotated[
+        Path,
+        typer.Option(
+            "--output-dir",
+            "-o",
+            help="Directory to save the prompts to.",
+            dir_okay=True,
+            file_okay=False,
+        ),
+    ] = Path("data/prompts"),
+    prompt_type: Annotated[
+        Literal["text", "chat"] | None,
+        typer.Option(
+            "--type", "-t", help="The type of prompt to pull or all if not specified."
+        ),
+    ] = None
+) -> None:
+    """
+    Pull prompt to filesystem from a given label and prompt type.
+
+    Examples:
+      bto_langfuse-cli prompt pull dev -o data/prompts -t chat -- → pull all chat prompts tagged dev to data/prompts.
+    """
+    langfuse = LangfuseService()
+    exporter = MarkdownPromptExporter()
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    written = 0
+    skipped = 0
+
+    typer.echo(f"Pulling prompts with label '{label}'...")
+
+    for meta, prompt_client in langfuse.get_prompts_with_meta(
+        label=label, prompt_type=prompt_type
+    ):
+        try:
+            out_path = exporter.export(meta, prompt_client, output_dir)
+            if out_path:
+                typer.echo(f"Wrote {out_path}")
+                written += 1
+        except Exception as e:
+            typer.echo(f"Error exporting {meta.name}: {e}", err=True)
+            skipped += 1
+
+    typer.echo(f"Done. Pulled {written} prompts to {output_dir} (Skipped: {skipped})")

src/bto_langfuse_cli/langfuse/langfuse_service.py

@@ -1,7 +1,11 @@
 import logging
+from typing import Iterator, Tuple
 
 from bto_langfuse_cli.langfuse.promote_plan import Plan
 from langfuse import get_client, Langfuse
+from langfuse.api.prompts.types.prompt_type import PromptType
+from langfuse.api.prompts.types.prompt_meta import PromptMeta
+from langfuse.model import TextPromptClient, ChatPromptClient
 
 logger = logging.getLogger(__name__)
 
@@ -17,18 +21,24 @@ def __init__(self):
             logger.error("Authentication failed. Please check your credentials and host.")
             raise Exception("Authentication failed. Please check your credentials and host.")
 
-    def get_all_prompts_names(self, label: str | None = None, tag: str | None = None) -> set[str]:
+    def get_all_prompts_names(
+        self, label: str | None = None, tag: str | None = None
+    ) -> set[str]:
         prompt_ids = set()
         current_page = 1
         limit = 50
 
-        first_response = self._langfuse.api.prompts.list(label=label, tag=tag, limit=limit, page=current_page)
+        first_response = self._langfuse.api.prompts.list(
+            label=label, tag=tag, limit=limit, page=current_page
+        )
         remaining_pages = first_response.meta.total_pages - current_page
         current_page = first_response.meta.page + 1
         prompt_ids.update({p.name for p in first_response.data})
 
         while remaining_pages > 0:
-            next_response = self._langfuse.api.prompts.list(label=label, tag=tag, limit=limit, page=current_page)
+            next_response = self._langfuse.api.prompts.list(
+                label=label, tag=tag, limit=limit, page=current_page
+            )
             remaining_pages = next_response.meta.total_pages - current_page
             current_page = next_response.meta.page + 1
             prompt_ids.update({p.name for p in next_response.data})
@@ -40,6 +50,49 @@ def _get_prompts(self, label: str) -> dict[str, int]:
         prompt_data = [self._langfuse.get_prompt(p, label=label) for p in prompts_names]
         return {p.name: p.version for p in prompt_data}
 
+    def get_prompts_with_meta(
+        self, label: str, prompt_type: str | None = None
+    ) -> Iterator[Tuple[PromptMeta, TextPromptClient | ChatPromptClient]]:
+        current_page = 1
+        limit = 50
+
+        while True:
+            response = self._langfuse.api.prompts.list(
+                label=label, limit=limit, page=current_page
+            )
+            metas = response.data
+
+            if not metas:
+                break
+
+            for meta in metas:
+                if prompt_type:
+                    expected_type = (
+                        PromptType.TEXT if prompt_type == "text" else PromptType.CHAT
+                    )
+                    if meta.type != expected_type:
+                        continue
+                elif meta.type not in (PromptType.TEXT, PromptType.CHAT):
+                    continue
+
+                langfuse_prompt_type = (
+                    "text" if meta.type is PromptType.TEXT else "chat"
+                )
+                try:
+                    prompt_client = self._langfuse.get_prompt(
+                        meta.name,
+                        label=label,
+                        type=langfuse_prompt_type,
+                        cache_ttl_seconds=0,
+                    )
+                    yield meta, prompt_client
+                except Exception as e:
+                    logger.error(f"Failed to fetch prompt {meta.name}: {e}")
+
+            if current_page >= response.meta.total_pages:
+                break
+            current_page += 1
+
     def plan(self, src_label, tgt_label) -> Plan:
         src_prompts = self._get_prompts(src_label)
         tgt_prompts = self._get_prompts(tgt_label)
@@ -82,4 +135,4 @@ def apply(self, plan: Plan, tgt_label: str):
             logger.info(f"Updating prompt {p.name} version {p.src_version} set labels {labels}")
             self._langfuse.update_prompt(name=p.name, version=p.src_version, new_labels=labels)
 
-        self._langfuse.flush()
+        self._langfuse.flush()

src/bto_langfuse_cli/langfuse/prompt_exporter.py

@@ -0,0 +1,123 @@
+from abc import ABC, abstractmethod
+from datetime import datetime
+from pathlib import Path
+
+from langfuse.api import PromptType
+from langfuse.api.prompts.types.prompt_meta import PromptMeta
+from langfuse.model import TextPromptClient, ChatPromptClient
+
+
+class PromptExporter(ABC):
+    """Interface for prompt exporters."""
+
+    @abstractmethod
+    def export(
+        self,
+        meta: PromptMeta,
+        prompt_client: TextPromptClient | ChatPromptClient,
+        output_dir: Path,
+    ) -> Path | None:
+        """
+        Export the prompt as a file to the given directory.
+
+        Args:
+            meta: The metadata of the prompt from the list API.
+            prompt_client: The detailed prompt client object.
+            output_dir: The directory where the prompt should be saved.
+
+        Returns:
+            The path to the generated file, or None if skipped.
+        """
+        pass
+
+
+class MarkdownPromptExporter(PromptExporter):
+    """
+    Exports prompts to Markdown files with YAML frontmatter
+    and XML section for Chat prompts.
+    """
+
+    def _yaml_quote(self, s: str) -> str:
+        if s == "":
+            return '""'
+        if any(c in s for c in (":", "#", "\n", '"')) or s.strip() != s:
+            escaped = s.replace("\\", "\\\\").replace('"', '\\"')
+            return f'"{escaped}"'
+        return s
+
+    def _format_yaml_block_list(self, key: str, items: list[str]) -> str:
+        if not items:
+            return f"{key}: []"
+        lines = [f"{key}:"]
+        for item in items:
+            lines.append(f"  - {self._yaml_quote(item)}")
+        return "\n".join(lines)
+
+    def _metadata_header(
+        self,
+        name: str,
+        version: int,
+        prompt_type: str,
+        labels: list[str],
+        tags: list[str],
+        last_updated_at: datetime,
+    ) -> str:
+        lines = [
+            "---",
+            f"name: {self._yaml_quote(name)}",
+            f"version: {version}",
+            f"type: {prompt_type}",
+            self._format_yaml_block_list("labels", labels),
+            self._format_yaml_block_list("tags", tags),
+            f"last_updated_at: {self._yaml_quote(last_updated_at.isoformat())}",
+            "---",
+        ]
+        return "\n".join(lines) + "\n"
+
+    def _serialize_chat_prompt(self, client: ChatPromptClient) -> str:
+        parts: list[str] = []
+        for msg in client.prompt:
+            if msg["type"] == "message":
+                role = msg["role"]
+                content = msg["content"]
+                parts.append(f"<{role}>\n\n{content}\n\n</{role}>")
+            elif msg["type"] == "placeholder":
+                name = msg["name"]
+                parts.append(f"<placeholder>{{{{{name}}}}}</placeholder>")
+            else:
+                raise ValueError(f"Unsupported chat message entry: {msg!r}")
+        return "\n\n".join(parts)
+
+    def export(
+        self,
+        meta: PromptMeta,
+        prompt_client: TextPromptClient | ChatPromptClient,
+        output_dir: Path,
+    ) -> Path | None:
+        out_path = Path(output_dir, meta.name).with_suffix(".md")
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+
+        langfuse_prompt_type = "text" if meta.type is PromptType.TEXT else "chat"
+
+        header = self._metadata_header(
+            name=prompt_client.name,
+            version=prompt_client.version,
+            prompt_type=langfuse_prompt_type,
+            labels=list(prompt_client.labels),
+            tags=list(prompt_client.tags),
+            last_updated_at=meta.last_updated_at,
+        )
+
+        match prompt_client:
+            case TextPromptClient():
+                body = prompt_client.prompt
+            case ChatPromptClient():
+                body = self._serialize_chat_prompt(prompt_client)
+            case _:
+                raise ValueError(
+                    f"Expected ChatPromptClient or TextPromptClient for {meta.name}"
+                )
+
+        combined = f"{header}\n{body}\n"
+        out_path.write_text(combined, encoding="utf-8")
+        return out_path