### Run Quickstart Flow Example

Source: https://github.com/hurtener/penguiflow/blob/main/docs/deployment/overview.md

Execute the quickstart flow example using the 'uv run' command.

```bash
uv run python examples/quickstart/flow.py
```

--------------------------------

### Penguiflow V2 Quick Start Guide

Source: https://github.com/hurtener/penguiflow/blob/main/examples/planner_enterprise_agent_v2/SUMMARY.md

Provides a quick 2-minute guide to setting up and running Penguiflow V2 with reflection enabled.

```bash
# 1. Navigate to v2 directory
cd examples/planner_enterprise_agent_v2

# 2. Copy environment template
cp .env.example .env

# 3. Set your API key
echo "OPENAI_API_KEY=sk-...". >> .env

# 4. Run with reflection enabled (default)
uv run python main.py

# 5. See reflection in action!
# The agent will critique its own answers before returning
```

--------------------------------

### Setup Development Environment

Source: https://github.com/hurtener/penguiflow/blob/main/CONTRIBUTING.md

Initialize the virtual environment and install development dependencies.

```bash
uv venv
uv pip install -e ".[dev]"
```

--------------------------------

### Build and serve documentation

Source: https://github.com/hurtener/penguiflow/blob/main/docs/contributing/dev-setup.md

Installs documentation dependencies and starts the local MkDocs development server.

```bash
uv pip install -e ".[dev,docs]"
uv run mkdocs serve
```

--------------------------------

### Complete Configuration Example

Source: https://github.com/hurtener/penguiflow/blob/main/docs/tools/configuration-guide.md

A full implementation showing registry setup, OAuth management, and multiple tool configurations.

```python
import os
from penguiflow.tools import (
    ToolNode,
    ExternalToolConfig,
    TransportType,
    AuthType,
    UtcpMode,
    RetryPolicy,
    OAuthManager,
    OAuthProviderConfig,
)
from penguiflow.registry import ModelRegistry

# Shared registry
registry = ModelRegistry()

# OAuth manager for user-level auth
oauth_manager = OAuthManager(
    providers={
        "github": OAuthProviderConfig(
            name="github",
            display_name="GitHub",
            auth_url="https://github.com/login/oauth/authorize",
            token_url="https://github.com/login/oauth/access_token",
            client_id=os.environ["GITHUB_CLIENT_ID"],
            client_secret=os.environ["GITHUB_CLIENT_SECRET"],
            redirect_uri="https://myapp.com/oauth/callback",
            scopes=["repo", "user"],
        ),
    },
)

# MCP server with OAuth
github = ToolNode(
    config=ExternalToolConfig(
        name="github",
        description="GitHub repository management",
        transport=TransportType.MCP,
        connection="npx -y @modelcontextprotocol/server-github",
        auth_type=AuthType.OAUTH2_USER,
        timeout_s=30.0,
        max_concurrency=10,
        tool_filter=["create_issue", "list_issues", "get_repo"],
    ),
    registry=registry,
    auth_manager=oauth_manager,
)

# UTCP API with bearer token
stripe = ToolNode(
    config=ExternalToolConfig(
        name="stripe",
        description="Stripe payment processing",
        transport=TransportType.UTCP,
        connection="https://stripe.com/.well-known/utcp.json",
        utcp_mode=UtcpMode.MANUAL_URL,
        auth_type=AuthType.BEARER,
        auth_config={"token": "${STRIPE_SECRET_KEY}"},
        timeout_s=45.0,
        max_concurrency=5,
        retry_policy=RetryPolicy(
            max_attempts=3,
            wait_exponential_min_s=0.5,
            wait_exponential_max_s=10.0,
            retry_on_status=[429, 500, 502, 503, 504],
        ),
    ),
    registry=registry,
)
```

--------------------------------

### Install MLflow and Run PenguiFlow Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/mlflow_metrics/README.md

Install the MLflow optional dependency using uv pip and then run the PenguiFlow example. This enables logging of metrics and tags to a local mlruns/ folder.

```bash
uv pip install mlflow
uv run python examples/mlflow_metrics/flow.py
```

--------------------------------

### Run Penguiflow examples

Source: https://github.com/hurtener/penguiflow/blob/main/docs/observability/telemetry-patterns.md

Execute the provided quickstart and roadmap status update flows using the uv package manager.

```bash
uv run python examples/quickstart/flow.py
uv run python examples/roadmap_status_updates/flow.py
```

--------------------------------

### Bootstrap and Generate Agent Workspace

Source: https://github.com/hurtener/penguiflow/blob/main/docs/cli/generate-command.md

This example shows the full workflow: initializing a new agent, navigating into its directory, editing the spec, generating the workspace with verbose output, and finally starting the development server.

```bash
uv run penguiflow generate --init my-agent
cd my-agent
# edit my-agent.yaml
uv run penguiflow generate --spec my-agent.yaml --verbose
uv run penguiflow dev --project-root .
```

--------------------------------

### Quickstart Flow Output Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/quickstart/README.md

The script output indicates the number of documents summarized, confirming successful message propagation and processing through the pipeline nodes.

```text
[metrics] summarize 2 docs
```

--------------------------------

### Run Visualizer Example

Source: https://github.com/hurtener/penguiflow/blob/main/docs/reference/visualization.md

Execute the runnable visualizer example from the repository.

```bash
uv run python examples/visualizer/flow.py
```

--------------------------------

### ReactPlanner Usage Example

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Example of initializing and running the ReactPlanner with a catalog of tools.

```APIDOC
## Initialize and Run ReactPlanner

### Description
This snippet demonstrates how to initialize the `ReactPlanner` with a language model and a catalog of tools, and then run it to process a query.

### Method
`planner.run(query)`

### Endpoint
N/A (Local execution)

### Request Example
```python
from penguiflow.planner import ReactPlanner
from penguiflow.tools import triage, retrieve, summarize

# Initialize the planner with an LLM and a catalog of tools
planner = ReactPlanner(llm="gpt-4", catalog=[triage, retrieve, summarize])

# Run the planner with a user query
result = await planner.run("Explain PenguiFlow")
```

### Response Example
```json
{
  "result": "[Explanation of PenguiFlow]"
}
```
```

--------------------------------

### Run Web Tools Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/web_tools/README.md

Executes the web tools flow example using uv.

```bash
uv run python examples/web_tools/flow.py
```

--------------------------------

### Execute the Routing Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/routing_with_playbooks/README.md

Run the routing example using the uv package manager.

```bash
uv run python examples/routing_with_playbooks/flow.py
```

--------------------------------

### Install and Configure Generated Agent

Source: https://github.com/hurtener/penguiflow/blob/main/docs/howto/uv-spec-workflow.md

Navigates into the generated agent directory, synchronizes its dependencies using uv, and copies the example environment file to `.env` for configuration.

```bash
cd echo-agent
uv sync
cp .env.example .env
```

--------------------------------

### Run the roadmap status subflows example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/roadmap_status_updates_subflows/README.md

Execute the example flow using the uv package manager.

```bash
uv run python examples/roadmap_status_updates_subflows/flow.py
```

--------------------------------

### Create ReactPlanner Agent with CLI

Source: https://github.com/hurtener/penguiflow/blob/main/docs/getting-started/quickstart.md

Use this command to quickly start a new agent project with the ReactPlanner template. Ensure 'uv' is installed and configured.

```bash
uv run penguiflow new my-agent --template react
```

--------------------------------

### Run Routing Examples

Source: https://github.com/hurtener/penguiflow/blob/main/docs/core/routers-and-policies.md

Command to run the provided routing examples.

```bash
uv run python examples/routing_predicate/flow.py
```

```bash
uv run examples/routing_union/flow.py
```

```bash
uv run examples/routing_policy/flow.py
```

--------------------------------

### Run Memory Basic Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/memory_basic/README.md

Execute the provided memory basic example script using uv.

```bash
uv run python examples/memory_basic/flow.py
```

--------------------------------

### Run the Discriminated Union Router example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/routing_union/README.md

Executes the routing example script using uv.

```bash
uv run python examples/routing_union/flow.py
```

--------------------------------

### Install PenguiFlow from Source

Source: https://github.com/hurtener/penguiflow/blob/main/docs/getting-started/installation.md

For contributors, clone the repository and install PenguiFlow in editable mode with development and documentation extras.

```bash
git clone git@github.com:hurtener/penguiflow.git
cd penguiflow
uv venv
uv pip install -e ".[dev,docs]"
```

--------------------------------

### Execute Predicate Router Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/routing_predicate/README.md

Run the predicate routing example using the uv package manager.

```bash
uv run python examples/routing_predicate/flow.py
```

--------------------------------

### Query Parameter Usage Examples

Source: https://github.com/hurtener/penguiflow/blob/main/docs/spec/a2a_specification.md

Examples of GET requests utilizing query parameters for filtering and history retrieval.

```http
GET /v1/tasks?contextId=uuid&status=working&pageSize=50&pageToken=cursor
```

```http
GET /v1/tasks/{id}?historyLength=10
```

--------------------------------

### Run A2A gRPC server example

Source: https://github.com/hurtener/penguiflow/blob/main/docs/deployment/distributed-execution.md

Use this command to run the A2A gRPC server example. Ensure you have the necessary dependencies installed.

```bash
uv run python examples/a2a_grpc_server/flow.py
```

--------------------------------

### Configure Environment Example

Source: https://github.com/hurtener/penguiflow/blob/main/TEMPLATING_QUICKGUIDE.md

Copies the example environment file to `.env`. Add provider API keys to `.env` if using a real model backend.

```bash
cp .env.example .env
# If you switched to a real model backend, add provider API keys to `.env`
```

--------------------------------

### Initiate a Long-Running Trace for Cancellation Example

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Example demonstrating how to start a flow and emit a message to initiate a long-running trace, which can then be targeted for cancellation.

```python
import asyncio
from penguiflow import create, Message, Headers

flow = create(slow_node.to(process_node))
flow.run()

# Start long-running trace
message = Message(payload="data", headers=Headers(tenant="user-123"))
await flow.emit(message)
```

--------------------------------

### Run the Prompt Injection Guardrail Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/guardrails/huggingface/README.md

Execute the example script for the HuggingFace prompt injection guardrail. If transformers is not installed, the guardrail will fail open.

```bash
uv run python examples/guardrails/huggingface/flow.py
```

--------------------------------

### Basic ReactPlanner Agent Setup and Execution

Source: https://context7.com/hurtener/penguiflow/llms.txt

Demonstrates setting up a ReactPlanner agent by registering tools, building a catalog, configuring the planner, and running it with a user query. Ensure all necessary imports are present before execution.

```python
import asyncio
import json
from collections.abc import Mapping
from typing import Any

from pydantic import BaseModel

from penguiflow.catalog import build_catalog, tool
from penguiflow.node import Node
from penguiflow.planner import ReactPlanner
from penguiflow.registry import ModelRegistry


class Question(BaseModel):
    text: str


class Intent(BaseModel):
    intent: str


class Documents(BaseModel):
    documents: list[str]


class FinalAnswer(BaseModel):
    answer: str


@tool(desc="Detect the caller intent", tags=["planner"])
async def triage(args: Question, ctx: object) -> Intent:
    return Intent(intent="docs")


@tool(desc="Retrieve supporting documents", side_effects="read")
async def retrieve(args: Intent, ctx: object) -> Documents:
    return Documents(documents=[f"PenguiFlow doc for {args.intent}"])


@tool(desc="Summarise retrieved documents")
async def summarise(args: FinalAnswer, ctx: object) -> FinalAnswer:
    return args


async def main() -> None:
    # Register node types
    registry = ModelRegistry()
    registry.register("triage", Question, Intent)
    registry.register("retrieve", Intent, Documents)
    registry.register("summarise", FinalAnswer, FinalAnswer)

    nodes = [
        Node(triage, name="triage"),
        Node(retrieve, name="retrieve"),
        Node(summarise, name="summarise"),
    ]

    # Build catalog from nodes
    catalog = build_catalog(nodes, registry)

    # Create planner with LLM
    planner = ReactPlanner(
        llm="gpt-4o",  # LiteLLM model identifier
        catalog=catalog,
        max_iters=10,
        temperature=0.0,
        token_budget=8000,  # Trigger summarization when exceeded
        hop_budget=20,      # Maximum tool calls
        deadline_s=120.0,   # Wall-clock timeout
    )

    # Run the planner
    result = await planner.run(
        "Explain PenguiFlow's architecture",
        llm_context={"user_preferences": {"verbosity": "concise"}},
        tool_context={"tenant_id": "acme", "user_id": "u123"},
    )

    print(f"Reason: {result.reason}")  # "answer_complete", "no_path", "budget_exhausted"
    print(f"Answer: {result.payload}")


if __name__ == "__main__":
    asyncio.run(main())

```

--------------------------------

### Run PenguiFlow Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/mlflow_metrics/README.md

Execute the PenguiFlow example script using uv. This command runs the flow without MLflow installed, printing metrics to stdout.

```bash
uv run python examples/mlflow_metrics/flow.py
```

--------------------------------

### Run Memory Persistence Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/memory_persistence/README.md

Execute the memory persistence example using the uv run command. This command starts the Python script located at examples/memory_persistence/flow.py.

```bash
uv run python examples/memory_persistence/flow.py
```

--------------------------------

### Run React Planner Minimal Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/react_minimal/README.md

Execute the React planner minimal example using the uv run command. Ensure you are in the correct directory.

```bash
uv run python examples/react_minimal/main.py
```

--------------------------------

### Runnable Example: Best-Effort Cancel

Source: https://github.com/hurtener/penguiflow/blob/main/docs/core/cancellation.md

A practical example demonstrating how to use the cancellation feature. It starts a long-running node, cancels the trace, and verifies that no final result is produced.

```APIDOC
## Runnable Example: Best-Effort Cancel

### Description
This example showcases the best-effort cancellation mechanism in PenguiFlow. It simulates a long-running task, cancels its trace, and confirms that the task is interrupted and no result is produced.

### Code Example
```python
from __future__ import annotations

import asyncio

from penguiflow import Headers, Message, Node, NodePolicy, create


async def slow(msg: Message, _ctx) -> None:
    del msg
    await asyncio.sleep(10.0)


slow_node = Node(slow, name="slow", policy=NodePolicy(validate="none"))


async def main() -> None:
    flow = create(slow_node.to())
    flow.run()

    message = Message(payload={"work": "x"}, headers=Headers(tenant="demo"))
    await flow.emit(message, trace_id=message.trace_id)

    cancelled = await flow.cancel(message.trace_id)
    print("cancelled:", cancelled)

    try:
        await asyncio.wait_for(flow.fetch(trace_id=message.trace_id), timeout=0.2)
        print("unexpected result")
    except asyncio.TimeoutError:
        print("no result (expected)")

    await flow.stop()


if __name__ == "__main__":
    asyncio.run(main())
```
```

--------------------------------

### Minimal Penguiflow Working Example

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

This example demonstrates the basic setup of a Penguiflow application, including defining Pydantic models, creating a Node, registering types with ModelRegistry, and running the flow.

```python
from penguiflow import create, Node, Message, Headers, ModelRegistry
from pydantic import BaseModel

# Define models
class Input(BaseModel):
    name: str

class Output(BaseModel):
    greeting: str

# Define node
async def greet(input: Input, ctx) -> Output:
    return Output(greeting=f"Hello, {input.name}!")

# Create flow
node = Node(greet, name="greet")
flow = create(node.to())

# Register types
registry = ModelRegistry()
registry.register("greet", Input, Output)

# Run
flow.run(registry=registry)
await flow.emit(Message(payload={"name": "World"}, headers=Headers(tenant="default")))
result = await flow.fetch()
print(result.payload)  # {"greeting": "Hello, World!"}
await flow.stop()
```

--------------------------------

### Minimal ReactPlanner Setup and Execution

Source: https://github.com/hurtener/penguiflow/blob/main/REACT_PLANNER_INTEGRATION_GUIDE.md

Demonstrates the basic setup for ReactPlanner, including defining tools, building a catalog, and running a simple query. It also shows how to extract structured payloads from the planner's result, as introduced in v2.7.

```python
import asyncio
from pydantic import BaseModel
from penguiflow.catalog import build_catalog, tool
from penguiflow.node import Node
from penguiflow.planner import ReactPlanner, ToolContext
from penguiflow.registry import ModelRegistry

# Define models
class Query(BaseModel):
    text: str

class Answer(BaseModel):
    response: str

# Define a tool
@tool(desc="Answer a question", tags=["answer"])
async def answer_query(args: Query, ctx: ToolContext) -> Answer:
    return Answer(response=f"Response to: {args.text}")

# Build and run
async def main():
    registry = ModelRegistry()
    registry.register("answer_query", Query, Answer)

    nodes = [Node(answer_query, name="answer_query")]
    catalog = build_catalog(nodes, registry)

    planner = ReactPlanner(
        llm="gpt-4o-mini",
        catalog=catalog,
        max_iters=4,
    )

    result = await planner.run("What is 2+2?")

    # Extract structured payload (v2.7+)
    payload = result.payload
    print(f"Answer: {payload['raw_answer']}")
    print(f"Confidence: {payload.get('confidence')}")
    print(f"Artifacts: {payload.get('artifacts', {})}")

asyncio.run(main())
```

--------------------------------

### Define Flow for Visualization

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Example setup of nodes and flow connections used for generating diagrams.

```python
controller_node = Node(controller, name="controller", allow_cycle=True)
summarize_node = Node(summarize, name="summarize")

flow = create(
    controller_node.to(controller_node, summarize_node),
    summarize_node.to()
)
```

--------------------------------

### Get PenguiFlow Version

Source: https://github.com/hurtener/penguiflow/blob/main/SUPPORT.md

Use this command to retrieve the installed PenguiFlow version, which is helpful when reporting issues.

```python
python -c "import penguiflow; print(penguiflow.__version__)"
```

--------------------------------

### Run ToolNode MCP Presets Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/toolnode_presets/README.md

Executes the example script to display preset configurations without calling external services.

```bash
uv run python examples/toolnode_presets/flow.py
```

--------------------------------

### Runnable Example: Best-Effort Cancel

Source: https://github.com/hurtener/penguiflow/blob/main/docs/core/cancellation.md

This example demonstrates best-effort cancellation by starting a long-running node, emitting a message, cancelling its trace, and verifying that no result is produced. It highlights the importance of cooperative nodes and proper cancellation handling.

```python
from __future__ import annotations

import asyncio

from penguiflow import Headers, Message, Node, NodePolicy, create


async def slow(msg: Message, _ctx) -> None:
    del msg
    await asyncio.sleep(10.0)


slow_node = Node(slow, name="slow", policy=NodePolicy(validate="none"))


async def main() -> None:
    flow = create(slow_node.to())
    flow.run()

    message = Message(payload={"work": "x"}, headers=Headers(tenant="demo"))
    await flow.emit(message, trace_id=message.trace_id)

    cancelled = await flow.cancel(message.trace_id)
    print("cancelled:", cancelled)

    try:
        await asyncio.wait_for(flow.fetch(trace_id=message.trace_id), timeout=0.2)
        print("unexpected result")
    except asyncio.TimeoutError:
        print("no result (expected)")

    await flow.stop()


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Run playbook examples via CLI

Source: https://github.com/hurtener/penguiflow/blob/main/docs/core/playbooks.md

Execute the provided playbook examples using the uv package manager.

```bash
uv run python examples/playbook_retrieval/flow.py
uv run python examples/routing_with_playbooks/flow.py
uv run python examples/roadmap_status_updates_subflows/flow.py
```

--------------------------------

### Partial Failure Handling

Source: https://github.com/hurtener/penguiflow/blob/main/docs/RFC/Done/RFC_TASK_GROUPS.md

Example of a task group setup where one task might fail, requiring a strategy for partial results.

```python
tasks.spawn(query="Analyze sales", group="q4")
tasks.spawn(query="Analyze marketing", group="q4")
tasks.spawn(query="Analyze ops", group="q4", group_sealed=True)

# Marketing task fails due to API error
```

--------------------------------

### Minimal Working Example

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Provides a minimal Python code example demonstrating how to set up and run the ReactPlanner with custom tools and data models.

```APIDOC
### 19.3 Basic Usage

#### Minimal Working Example

```python
from penguiflow.catalog import build_catalog, tool
from penguiflow.node import Node
from penguiflow.planner import ReactPlanner
from penguiflow.registry import ModelRegistry
from pydantic import BaseModel

# 1. Define data models
class Question(BaseModel):
    text: str

class Intent(BaseModel):
    intent: str

class Documents(BaseModel):
    documents: list[str]

# 2. Define tools with @tool decorator
@tool(desc="Detect the caller intent", tags=["triage"])
async def triage(args: Question, ctx: object) -> Intent:
    intent = "docs" if "explain" in args.text.lower() else "general"
    return Intent(intent=intent)

@tool(desc="Retrieve supporting documents", side_effects="read")
async def retrieve(args: Intent, ctx: object) -> Documents:
    docs = [f"Content about {args.intent}"]
    return Documents(documents=docs)

# 3. Setup registry
registry = ModelRegistry()
registry.register("triage", Question, Intent)
registry.register("retrieve", Intent, Documents)

# 4. Create nodes
nodes = [
    Node(triage, name="triage"),
    Node(retrieve, name="retrieve"),
]

# 5. Initialize planner
planner = ReactPlanner(
    llm="gpt-4",  # or any LiteLLM-compatible model
    catalog=build_catalog(nodes, registry),
    max_iters=8,
    temperature=0.0,
)

# 6. Run
result = await planner.run("Explain PenguiFlow")

# 7. Handle result
if result.reason == "answer_complete":
    print(result.payload)
elif result.reason == "budget_exhausted":
    print("Hit iteration limit")
```
```

--------------------------------

### ReactPlanner Initialization with Planning Hints

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Demonstrates how to initialize the ReactPlanner with a comprehensive set of planning hints.

```APIDOC
## ReactPlanner Initialization with Planning Hints

### Description
Initializes the ReactPlanner with various planning hints to guide the LLM's execution strategy.

### Method
Instantiation

### Endpoint
N/A

### Parameters
#### Request Body
- **llm** (string) - Required - The language model to use.
- **catalog** (object) - Required - The tool catalog.
- **planning_hints** (object) - Optional - Configuration for planning hints.
  - **ordering_hints** (array of strings) - Optional - Suggests preferred tool sequence.
  - **parallel_groups** (array of arrays of strings) - Optional - Suggests which tools can run concurrently.
  - **sequential_only** (array of strings) - Optional - Tools that must never be parallelized.
  - **max_parallel** (integer) - Optional - Maximum concurrent branches allowed.
  - **disallow_nodes** (array of strings) - Optional - Tools that cannot be used.
  - **prefer_nodes** (array of strings) - Optional - Tools to prefer over alternatives.
  - **budget_hints** (object) - Optional - Budget-related hints.
    - **max_parallel** (integer) - Optional - Maximum parallel operations within budget.
    - **prefer_sequential** (boolean) - Optional - Preference for sequential execution.

### Request Example
```json
{
  "llm": "gpt-4",
  "catalog": catalog,
  "planning_hints": {
    "ordering_hints": ["triage", "retrieve", "summarize"],
    "parallel_groups": [["shard_0", "shard_1", "shard_2"]],
    "sequential_only": ["approval_gate", "write_database"],
    "max_parallel": 4,
    "disallow_nodes": ["deprecated_tool", "dangerous_operation"],
    "prefer_nodes": ["fast_search", "cached_retrieval"],
    "budget_hints": {
      "max_parallel": 3,
      "prefer_sequential": true
    }
  }
}
```

### Response
N/A (Object instantiation)
```

--------------------------------

### Run PenguiFlow Development Server with CLI

Source: https://github.com/hurtener/penguiflow/blob/main/docs/getting-started/quickstart.md

Use this command to start the development server for your PenguiFlow project. Ensure 'uv' is installed and configured.

```bash
uv run penguiflow dev --project-root my-agent
```

--------------------------------

### Minimal PenguiFlow ReactPlanner Example

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

A basic example demonstrating how to set up tools, nodes, and a ReactPlanner to process a user query. Ensure all necessary models and tools are registered.

```python
from penguiflow.catalog import build_catalog, tool
from penguiflow.node import Node
from penguiflow.planner import ReactPlanner
from penguiflow.registry import ModelRegistry
from pydantic import BaseModel

# 1. Define data models
class Question(BaseModel):
    text: str

class Intent(BaseModel):
    intent: str

class Documents(BaseModel):
    documents: list[str]

# 2. Define tools with @tool decorator
@tool(desc="Detect the caller intent", tags=["triage"])
async def triage(args: Question, ctx: object) -> Intent:
    intent = "docs" if "explain" in args.text.lower() else "general"
    return Intent(intent=intent)

@tool(desc="Retrieve supporting documents", side_effects="read")
async def retrieve(args: Intent, ctx: object) -> Documents:
    docs = [f"Content about {args.intent}"]
    return Documents(documents=docs)

# 3. Setup registry
registry = ModelRegistry()
registry.register("triage", Question, Intent)
registry.register("retrieve", Intent, Documents)

# 4. Create nodes
nodes = [
    Node(triage, name="triage"),
    Node(retrieve, name="retrieve"),
]

# 5. Initialize planner
planner = ReactPlanner(
    llm="gpt-4",  # or any LiteLLM-compatible model
    catalog=build_catalog(nodes, registry),
    max_iters=8,
    temperature=0.0,
)

# 6. Run
result = await planner.run("Explain PenguiFlow")

# 7. Handle result
if result.reason == "answer_complete":
    print(result.payload)
elif result.reason == "budget_exhausted":
    print("Hit iteration limit")
```

--------------------------------

### Initialize and Run ReactPlanner

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Instantiate ReactPlanner with a language model and a catalog of tools. Then, run the planner with a user query to get a result.

```python
planner = ReactPlanner(llm="gpt-4", catalog=[triage, retrieve, summarize])
result = await planner.run("Explain PenguiFlow")
```

--------------------------------

### Runnable Penguiflow Example

Source: https://github.com/hurtener/penguiflow/blob/main/docs/core/streaming.md

This example demonstrates creating a Penguiflow with multiple nodes for chunk processing and final answer delivery. Ensure all nodes are correctly defined and connected. Use `flow.run()` to start the flow and `flow.emit()` to send messages. `flow.fetch()` is used to retrieve the final result.

```python
from __future__ import annotations

import asyncio

from penguiflow import Headers, Message, Node, NodePolicy, create
from penguiflow.types import FinalAnswer


async def chunk_sink(msg: Message, _ctx) -> None:
    chunk = msg.payload
    print(chunk.text, end="")
    if chunk.done:
        print("")


async def deliver_final(msg: Message, _ctx) -> FinalAnswer:
    return msg.payload


async def compose(msg: Message, ctx) -> None:
    await ctx.emit_chunk(parent=msg, text="hello ", to=chunk_node)
    await ctx.emit_chunk(parent=msg, text="world", done=True, to=chunk_node)

    final = msg.model_copy(update={"payload": FinalAnswer(text="hello world")})
    await ctx.emit(final, to=final_node)


chunk_node = Node(chunk_sink, name="chunk_sink", policy=NodePolicy(validate="none"))
final_node = Node(deliver_final, name="final", policy=NodePolicy(validate="none"))
compose_node = Node(compose, name="compose", policy=NodePolicy(validate="none"))


async def main() -> None:
    flow = create(
        compose_node.to(chunk_node, final_node),
        chunk_node.to(),
        final_node.to(),
    )
    flow.run()

    message = Message(payload={"request": "ignored"}, headers=Headers(tenant="demo"))
    await flow.emit(message, trace_id=message.trace_id)
    result = await flow.fetch(trace_id=message.trace_id)
    print("Final:", result.text)

    await flow.stop()


if __name__ == "__main__":
    asyncio.run(main())

```

--------------------------------

### Run PenguiFlow Quickstart Flow

Source: https://github.com/hurtener/penguiflow/blob/main/examples/quickstart/README.md

Execute the quickstart flow using the `uv` command or directly with Python if the virtual environment is active.

```bash
uv run python examples/quickstart/flow.py
```

```bash
python examples/quickstart/flow.py
```

--------------------------------

### Run Parallel Join Demo

Source: https://github.com/hurtener/penguiflow/blob/main/examples/react_parallel_join/README.md

Execute the parallel join demonstration script using uv run. This command starts the example that fans out planner actions across providers.

```bash
uv run python examples/react_parallel_join/main.py
```

--------------------------------

### Complete PenguiFlow Pipeline Example

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Defines data models, node functions, registers types, builds a flow graph, and runs a message through the pipeline. Use this as a starting point for your first PenguiFlow application.

```python
import asyncio
from pydantic import BaseModel
from penguiflow import create, Node, NodePolicy, ModelRegistry, Message, Headers

# 1. Define your data models
class QueryIn(BaseModel):
    text: str

class TriageOut(BaseModel):
    text: str
    topic: str

class RetrieveOut(BaseModel):
    topic: str
    docs: list[str]

class PackOut(BaseModel):
    prompt: str

# 2. Define node functions
async def triage(msg: QueryIn, ctx) -> TriageOut:
    """Classify the query into a topic."""
    topic = "metrics" if "metric" in msg.text else "general"
    return TriageOut(text=msg.text, topic=topic)

async def retrieve(msg: TriageOut, ctx) -> RetrieveOut:
    """Fetch relevant documents for the topic."""
    docs = [f"doc_{i}_{msg.topic}" for i in range(2)]
    return RetrieveOut(topic=msg.topic, docs=docs)

async def pack(msg: RetrieveOut, ctx) -> PackOut:
    """Package documents into a prompt."""
    prompt = f"[{msg.topic}] summarize {len(msg.docs)} docs"
    return PackOut(prompt=prompt)

# 3. Wrap functions in Node objects
triage_node = Node(triage, name="triage", policy=NodePolicy(validate="both"))
retrieve_node = Node(retrieve, name="retrieve", policy=NodePolicy(validate="both"))
pack_node = Node(pack, name="pack", policy=NodePolicy(validate="both"))

# 4. Register input/output types for validation
registry = ModelRegistry()
registry.register("triage", QueryIn, TriageOut)
registry.register("retrieve", TriageOut, RetrieveOut)
registry.register("pack", RetrieveOut, PackOut)

# 5. Build the flow graph
flow = create(
    triage_node.to(retrieve_node),  # triage → retrieve
    retrieve_node.to(pack_node),    # retrieve → pack
)

# 6. Start the flow (non-blocking)
flow.run(registry=registry)

# 7. Emit a message to OpenSea
message = Message(
    payload=QueryIn(text="show marketing metrics"),
    headers=Headers(tenant="acme"),
)
await flow.emit(message)

# If multiple concurrent callers share the same running flow instance,
# use trace-scoped roundtrips to avoid cross-consuming results.
# (Pick a unique trace_id per request.)
# trace_id = "..."
# await flow.emit(message, trace_id=trace_id)
# result = await flow.fetch(trace_id=trace_id)

# 8. Fetch the result from Rookery
result = await flow.fetch()  # Returns: Message(payload=PackOut(...))
print(result.payload.prompt)  # "[metrics] summarize 2 docs"

# 9. Stop the flow
await flow.stop()
```

--------------------------------

### Run the Routing Policy Example

Source: https://github.com/hurtener/penguiflow/blob/main/examples/routing_policy/README.md

Executes the routing policy demonstration script using the uv package manager.

```bash
uv run python examples/routing_policy/flow.py
```

--------------------------------

### Initialize Project with Templates

Source: https://github.com/hurtener/penguiflow/blob/main/TEMPLATING_QUICKGUIDE.md

Commands to create new projects using specific architectural templates.

```bash
penguiflow new NAME --template=minimal
```

```bash
penguiflow new NAME
```

```bash
penguiflow new NAME --template=parallel
```

```bash
penguiflow new NAME --template=flow
```

```bash
penguiflow new NAME --template=controller
```

```bash
penguiflow new NAME --template=rag_server
```

```bash
penguiflow new NAME --template=wayfinder
```

```bash
penguiflow new NAME --template=analyst
```

```bash
penguiflow new NAME --template=enterprise
```

--------------------------------

### Create Repo and uv Environment

Source: https://github.com/hurtener/penguiflow/blob/main/docs/howto/uv-spec-workflow.md

Initializes a new directory, sets up Git, and creates a Python 3.11 environment using uv.

```bash
mkdir pf-workspace && cd pf-workspace
git init
uv init --python 3.11
```

--------------------------------

### Generate and Run React Agent

Source: https://github.com/hurtener/penguiflow/blob/main/TEMPLATING_QUICKGUIDE.md

Commands to generate and run a React template agent. This includes installing dependencies, copying environment files, and starting the agent. Remember to configure your API key in the `.env` file.

```bash
penguiflow generate --spec=hello_world.yaml
cd hello-agent
uv sync && cp .env.example .env
# Edit .env to add your API key
uv run python -m hello_agent

```

--------------------------------

### Test List Artifacts: No Session ID Provided

Source: https://github.com/hurtener/penguiflow/blob/main/docs/RFC/ToDo/issue-74/003-playground-hydration/phases/phase-002.md

Checks that if no session ID is provided via query parameters or headers, the GET /artifacts endpoint returns an empty list. This setup involves creating an in-memory artifact store and a mock agent.

```python
def test_list_artifacts_no_session_returns_empty(self, tmp_path: Path) -> None:
        """When no session ID is provided (no query param, no header), returns []."""
        store = InMemoryArtifactStore()
        wrapper = MockAgentWrapper()
        wrapper._planner = MagicMock()
        wrapper._planner.artifact_store = store

        app = create_playground_app(project_root=tmp_path, agent=wrapper)
        client = TestClient(app, raise_server_exceptions=False)

        response = client.get("/artifacts")
        assert response.status_code == 200
        assert response.json() == []
```

--------------------------------

### Initialize a minimal ReactPlanner instance

Source: https://github.com/hurtener/penguiflow/blob/main/docs/planner/configuration.md

Demonstrates a basic planner setup without memory or steering, using a simple echo tool and a model registry.

```python
from __future__ import annotations

from pydantic import BaseModel

from penguiflow import ModelRegistry, Node
from penguiflow.catalog import build_catalog, tool
from penguiflow.planner import ReactPlanner, ToolContext


class EchoArgs(BaseModel):
    payload: dict


class EchoOut(BaseModel):
    payload: dict


@tool(desc="Echo input", side_effects="pure")
async def echo(args: EchoArgs, ctx: ToolContext) -> EchoOut:
    del ctx
    return EchoOut(payload=args.payload)


def build_planner() -> ReactPlanner:
    registry = ModelRegistry()
    registry.register("echo", EchoArgs, EchoOut)
    catalog = build_catalog([Node(echo, name="echo")], registry)
    return ReactPlanner(llm="gpt-4o-mini", catalog=catalog)
```

--------------------------------

### Custom PenguiFlow Benchmark Script

Source: https://github.com/hurtener/penguiflow/blob/main/manual.md

Create custom benchmark scripts using PenguiFlow's API to measure latency, throughput, and memory usage. This example demonstrates starting memory profiling, running a flow, and calculating statistics over multiple message emissions.

```python
import time
import tracemalloc
from penguiflow import create, Node, Message, NodePolicy

# Start memory profiling
tracemalloc.start()

# Build flow with specific configuration
flow = create(
    node_a.to(node_b),
    node_b.to(node_c),
    queue_maxsize=128  # Test configuration
)
flow.run()

# Measure latency over N iterations
latencies = []
start_wall = time.perf_counter()

for i in range(1000):
    message = Message(payload=f"msg-{i}")

    start = time.perf_counter()
    await flow.emit(message)
    result = await flow.fetch()
    elapsed_ms = (time.perf_counter() - start) * 1000

    latencies.append(elapsed_ms)

wall_time = time.perf_counter() - start_wall

# Calculate statistics
import statistics
print(f"Average latency: {statistics.mean(latencies):.2f}ms")
print(f"p50: {statistics.median(latencies):.2f}ms")
print(f"p90: {sorted(latencies)[int(len(latencies)*0.9)]:.2f}ms")
print(f"Throughput: {1000/wall_time:.1f} messages/second")
```

--------------------------------

### Manage Projects via CLI

Source: https://context7.com/hurtener/penguiflow/llms.txt

Bootstrap projects, initialize tooling, and manage development servers or external tool connections.

```bash
# Create a new agent project from template
penguiflow new my-agent --template react

# Initialize VS Code tooling in current directory
penguiflow init

# Run the development playground server
penguiflow dev --project-root . --port 8001

# Connect to and test external tools
penguiflow tools connect github --transport mcp
penguiflow tools list github
```

--------------------------------

### Install Optional Dependencies

Source: https://github.com/hurtener/penguiflow/blob/main/examples/guardrails/huggingface/README.md

Install the necessary libraries for the HuggingFace guardrail. Ensure you have `transformers` and `torch` installed.

```bash
uv pip install transformers torch
```

--------------------------------

### Install dependencies

Source: https://github.com/hurtener/penguiflow/blob/main/docs/cli/new-command.md

Required installation commands for the scaffolding tool.

```bash
pip install "penguiflow[cli]"
```

```bash
pip install jinja2
```

--------------------------------

### Install PenguiFlow

Source: https://context7.com/hurtener/penguiflow/llms.txt

Standard installation command for the core library.

```bash
pip install penguiflow
```

--------------------------------

### Initialize and Run a Basic Penguiflow

Source: https://github.com/hurtener/penguiflow/blob/main/docs/migration/penguiflow-adoption.md

This snippet shows the basic setup for initializing a database pool, creating a flow, and running it with a registry. It demonstrates emitting a message and fetching results within a try-finally block to ensure the flow is stopped.

```python
db_pool = DatabasePool("postgresql://localhost/mydb")
await db_pool.initialize()

flow = create_flow_with_db(db_pool)
flow.run(registry=registry)

try:
    await flow.emit(Message(payload=input_data))
    result = await flow.fetch()
finally:
    await flow.stop()
```