#
# Copyright (c) 2025-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""``pipecat init`` — the single starting point for building a Pipecat app.
``pipecat init`` initializes a new Pipecat project, then routes you to a build method:
- ``AGENTS.md`` — the coding-agent guide (read natively by most coding agents).
- ``CLAUDE.md`` — a one-line ``@AGENTS.md`` import so Claude Code loads it too.
- ``GETTING_STARTED.md`` — for the *developer*: how to drive the agent well
(MCP setup, how to write the first prompt, what to expect in a session).
Deliberately not in CLAUDE.md/AGENTS.md — those cost agent context every
session, and this guidance is for the human. Written only on the coding-agent
path: it's onboarding for commissioning a build, so it doesn't fit a project
that's already been scaffolded (where the README is the start-here).
``init`` is also the scaffolder. After writing AGENTS.md + CLAUDE.md it either:
- builds the project from flags or a config file when any scaffold option is given
(``pipecat init . --bot-type web -t daily …``) — the non-interactive path coding
agents and automation use;
- or, interactively, asks how you want to build: hand off to a coding agent, or scaffold
a runnable bot right now (the wizard, in-place in the same directory).
``pipecat init quickstart`` skips the question and scaffolds the canned quickstart bot
in-place — the coding-agent guide *and* a runnable bot in one step. The scaffolding
itself lives in :mod:`pipecat.cli.scaffold`.
Editing policy for the bundled guide: keep API specifics (signatures, imports,
parameter names) out of AGENTS.md — it is a static snapshot, so anything that
churns belongs in the live sources the guide's §3 points agents at.
"""
import re
import sys
from pathlib import Path
import typer
from rich.console import Console
import pipecat.cli
from pipecat.cli.scaffold import (
generate_scaffold,
list_options_callback,
resolve_scaffold_config,
)
console = Console()
# Directory holding the bundled AGENTS.md / CLAUDE.md (shipped as package data; see
# pyproject [tool.setuptools.package-data] "pipecat.cli" and MANIFEST.in).
_AGENT_TEMPLATES = Path(pipecat.cli.__file__).parent / "agent_templates"
_AGENTS_FILE = "AGENTS.md"
_CLAUDE_FILE = "CLAUDE.md"
_GETTING_STARTED_FILE = "GETTING_STARTED.md"
# Fixed destination for `pipecat init quickstart`.
_QUICKSTART_DIR = "pipecat-quickstart"
# `--help` panels grouping the scaffold flags, so `pipecat init --help` reads as
# "what to build" vs. "optional features" instead of one flat wall of options.
_PANEL_SCAFFOLD = "Scaffold options"
_PANEL_FEATURES = "Scaffold features"
# Matches the version recorded in a guide footer, e.g. "(pipecat-ai 0.1.2)".
_FOOTER_VERSION_RE = re.compile(r"pipecat-ai ([^)]+)\)")
def _guide_footer() -> str:
"""Provenance stamp appended to pipecat-owned guide files.
The written guide is a static snapshot that otherwise looks like hand-written project
docs. The footer tells a later reader (human or agent) that it is generated and
refreshable, and which pipecat-ai wrote it — so a re-run can spot a stale guide and
nudge the developer to refresh it.
"""
return (
f"\n<!-- Generated by `pipecat init` (pipecat-ai {pipecat.__version__}). "
"Run `pipecat init --overwrite-guide` to refresh. -->\n"
)
def _stamped_version(text: str) -> str | None:
"""Return the pipecat-ai version recorded in a guide's footer, or None if unstamped."""
match = _FOOTER_VERSION_RE.search(text)
return match.group(1) if match else None
def _write_guide_file(path: Path, content: str, *, stamped: bool, overwrite: bool) -> str:
"""Write one guide file under the uniform preserve-by-default rule.
Absent → write it. Present → keep it (never clobber silently) unless ``overwrite``.
``stamped`` appends the provenance footer, so the file can be recognized and
version-checked on a later run; pass it for pipecat-owned guides (AGENTS.md,
GETTING_STARTED.md), not the trivial ``@AGENTS.md`` pointer in CLAUDE.md.
Prints a per-file line and returns a status — ``"wrote"``, ``"overwrote"``,
``"kept"``, or ``"kept-stale"`` (kept, but written by a different pipecat-ai) — that
the caller folds into a single refresh summary. Raises ``OSError`` on a write failure.
"""
body = content + _guide_footer() if stamped else content
if not path.exists():
path.write_text(body, encoding="utf-8")
console.print(f"[green]✔[/green] Wrote {path}")
return "wrote"
if overwrite:
path.write_text(body, encoding="utf-8")
console.print(f"[green]✔[/green] Overwrote {path}")
return "overwrote"
# Keep the existing file. Surface whether it's a stale pipecat guide (worth a refresh
# nudge) or one we don't own, so the developer knows why it was left alone.
stamped_ver = _stamped_version(path.read_text(encoding="utf-8"))
if stamped_ver is not None and stamped_ver != pipecat.__version__:
console.print(f"[yellow]•[/yellow] Kept {path} (written by pipecat-ai {stamped_ver})")
return "kept-stale"
if stamped and stamped_ver is None:
console.print(f"[yellow]•[/yellow] Kept {path} (your own file)")
return "kept"
console.print(f"[yellow]•[/yellow] Kept {path}")
return "kept"
def _print_refresh_summary(statuses: list[str]) -> None:
"""Point the developer at --overwrite-guide when a re-run changed nothing.
Keeps a preserve-everything run from reading as a silent no-op, and surfaces a stale
guide instead of letting it sit out of date quietly.
"""
if any(s == "kept-stale" for s in statuses):
console.print(
"[yellow]⚠[/yellow] Your guide was written by an older Pipecat. Run "
"[bold]`pipecat init --overwrite-guide`[/bold] to update it."
)
elif statuses and all(s.startswith("kept") for s in statuses):
console.print(
"Guide files already in place. Run "
"[bold]`pipecat init --overwrite-guide`[/bold] to refresh them."
)
def _offer_refresh(target_dir: Path) -> bool:
"""Stale guide + interactive terminal: offer to refresh it now instead of just warning.
Someone re-running ``init`` on a project whose guide predates their current Pipecat
almost always wants the new guide, so turn the passive nudge into a one-keystroke
action. Returns True if the guide was refreshed (the caller then refreshes the rest of
the guide too). Non-interactive callers don't reach here — they get the printed nudge.
"""
import questionary
from pipecat.cli.prompts.questions import custom_style
refresh = questionary.confirm(
"A guide file is from an older Pipecat. Refresh the guide files now?",
default=True,
style=custom_style,
).ask()
# `ask()` returns None on Ctrl-C / EOF — treat anything but an explicit yes as "keep".
if refresh:
_write_agent_guide(target_dir, overwrite=True)
return True
return False
def _write_agent_guide(target_dir: Path, overwrite: bool) -> list[str]:
"""Write the core agent guide — AGENTS.md and CLAUDE.md — into a directory.
Wanted on *every* path (coding agent, scaffold-now, quickstart), so they're written
upfront. Both follow the uniform rule (:func:`_write_guide_file`): an existing file is
kept unless ``overwrite``. AGENTS.md carries a provenance footer (so a re-run can spot
a stale guide); CLAUDE.md is the trivial ``@AGENTS.md`` pointer and is left unstamped.
Returns the per-file statuses so the caller can follow up — a passive refresh nudge,
or an interactive offer to refresh. The developer guide (``GETTING_STARTED.md``) is
written separately, by :func:`_write_developer_guide`, only on the coding-agent path.
Exits non-zero on a read/write error.
"""
try:
agents_src = (_AGENT_TEMPLATES / _AGENTS_FILE).read_text(encoding="utf-8")
claude_src = (_AGENT_TEMPLATES / _CLAUDE_FILE).read_text(encoding="utf-8")
except OSError as e: # bundled data missing — a packaging regression
console.print(f"[red]Error: bundled agent guide not found:[/red] {e}")
raise typer.Exit(1)
try:
target_dir.mkdir(parents=True, exist_ok=True)
return [
_write_guide_file(
target_dir / _AGENTS_FILE, agents_src, stamped=True, overwrite=overwrite
),
_write_guide_file(
target_dir / _CLAUDE_FILE, claude_src, stamped=False, overwrite=overwrite
),
]
except OSError as e:
console.print(f"[red]Error writing files:[/red] {e}")
raise typer.Exit(1)
def _write_developer_guide(target_dir: Path, overwrite: bool) -> None:
"""Write GETTING_STARTED.md — from-scratch developer onboarding — into a directory.
Written only on the coding-agent path, where the developer is about to commission a
build. It does *not* fit a project that's just been scaffolded (``Scaffold a runnable
bot now`` or ``pipecat init quickstart``): there the bot already exists and the
scaffold's README is the start-here. Follows the same preserve-by-default rule as the
rest of the guide; exits non-zero on a read/write error.
"""
try:
src = (_AGENT_TEMPLATES / _GETTING_STARTED_FILE).read_text(encoding="utf-8")
except OSError as e: # bundled data missing — a packaging regression
console.print(f"[red]Error: bundled agent guide not found:[/red] {e}")
raise typer.Exit(1)
try:
target_dir.mkdir(parents=True, exist_ok=True)
_write_guide_file(
target_dir / _GETTING_STARTED_FILE, src, stamped=True, overwrite=overwrite
)
except OSError as e:
console.print(f"[red]Error writing files:[/red] {e}")
raise typer.Exit(1)
def _is_interactive() -> bool:
"""Whether we can show an interactive prompt (a real terminal on both ends)."""
return sys.stdin.isatty() and sys.stdout.isatty()
def _print_ready(target_dir: Path) -> None:
"""Print the guidance shown when the developer keeps the coding-agent path.
Names the project directory so the developer knows where to open their session,
except when init targeted the current directory (``pipecat init .``), where "here"
reads better than "in .". The agent runs ``pipecat init`` itself to scaffold (per
AGENTS.md), so that's left out of this human-facing message.
"""
where = "here" if target_dir.resolve() == Path.cwd() else f"in [bold]{target_dir}[/bold]"
console.print(
f"\n[bold]Your project is ready.[/bold]\n\n"
f"Read [bold]{_GETTING_STARTED_FILE}[/bold] for how to prompt your agent, then open a "
f"coding session {where} to start building."
)
def _route_build_method(target_dir: Path, overwrite_guide: bool) -> None:
"""Ask whether to build with a coding agent or scaffold a runnable bot now.
The coding-agent path adds the developer guide (GETTING_STARTED.md) and points the
developer at their coding session. It's also the default when there's no interactive
terminal (automation, tests, piped input) or when re-running in an already-scaffolded
project to refresh the guide. The scaffold path runs the `create` wizard in-place and
skips the developer guide — the scaffold's README is the start-here for a built bot.
"""
already_scaffolded = (target_dir / "server").exists()
if already_scaffolded or not _is_interactive():
_write_developer_guide(target_dir, overwrite_guide)
_print_ready(target_dir)
return
import questionary
from pipecat.cli.prompts.questions import custom_style
choice = questionary.select(
"How do you want to build?",
choices=[
questionary.Choice("Build with a coding agent (recommended)", value="agent"),
questionary.Choice("Scaffold a runnable bot now", value="scaffold"),
],
style=custom_style,
).ask()
# `ask()` returns None on Ctrl-C / EOF — fall through to the safe agent path.
if choice != "scaffold":
_write_developer_guide(target_dir, overwrite_guide)
_print_ready(target_dir)
return
# Scaffold now: run the wizard in-place. The scaffold lands in the same directory that
# already holds AGENTS.md/CLAUDE.md (in-place mode preserves CLAUDE.md), and its README
# carries the start-here guidance — so no GETTING_STARTED.md here.
from pipecat.cli.scaffold import scaffold_interactive
try:
scaffold_interactive(target_dir, target_dir.name, in_place=True)
except KeyboardInterrupt:
console.print("\n[yellow]Project creation cancelled.[/yellow]")
raise typer.Exit(1)
except typer.Exit:
raise
except FileExistsError as e:
console.print(f"\n[red]Error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"\n[red]Error creating project: {e}[/red]")
raise typer.Exit(1)
def _init_quickstart(overwrite_guide: bool) -> None:
"""``pipecat init quickstart``: the canned quickstart, with the coding-agent guide.
Writes the agent guide into ``pipecat-quickstart/`` and scaffolds the quickstart
bot in-place there, so the learner's first project is both runnable and set up for
coding agents. Non-interactive — it's a fixed preset, so there's no build-method
question.
Skips ``GETTING_STARTED.md`` (the from-scratch developer onboarding): the bot
already exists, so the scaffold's README is the start-here.
"""
from pipecat.cli.scaffold import scaffold_quickstart
target_dir = Path(_QUICKSTART_DIR)
# AGENTS.md + CLAUDE.md only — _write_agent_guide never writes GETTING_STARTED.md.
# Quickstart is a fixed, non-interactive preset, so a stale guide just gets the nudge.
_print_refresh_summary(_write_agent_guide(target_dir, overwrite_guide))
try:
scaffold_quickstart(dest=target_dir, in_place=True)
except typer.Exit:
raise
except FileExistsError as e:
console.print(f"\n[red]Error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"\n[red]Error creating project: {e}[/red]")
raise typer.Exit(1)
# Scaffold options that, when present, switch `init` to non-interactive scaffolding.
# A truthy/non-None value on any of these (or `--config`) means "build the project now"
# rather than just writing the guide and prompting.
def _scaffold_requested(*, config, name, bot_type, transport, mode, stt, llm, tts, realtime, video):
return (
config is not None
or bool(transport)
or any(v is not None for v in (name, bot_type, mode, stt, llm, tts, realtime, video))
)
# Every scaffold/feature option, by parameter name — used to detect flags the user
# explicitly typed (vs. defaulted) so `init quickstart` can flag the ones it ignores.
_SCAFFOLD_PARAM_NAMES = (
"name", "bot_type", "transport", "mode", "stt", "llm", "tts", "realtime", "video",
"client_framework", "client_server", "daily_pstn_mode", "twilio_daily_sip_mode", "config",
"recording", "transcription", "video_input", "video_output", "deploy_to_cloud",
"enable_krisp", "observability", "enable_eval",
) # fmt: skip
def _has_explicit_scaffold_flags(ctx: typer.Context) -> bool:
"""Whether the user typed any scaffold/feature flag on the command line.
Uses the Typer context's parameter source so boolean toggles (which can't be told from
their default by value) are detected too.
"""
return any(
(src := ctx.get_parameter_source(p)) is not None and src.name == "COMMANDLINE"
for p in _SCAFFOLD_PARAM_NAMES
)
[docs]
def init_command(
ctx: typer.Context,
target: str | None = typer.Argument(
None,
help="Project directory: initialized, and scaffolded in place when scaffold options "
"are given. Use '.' for the current directory; created if missing. Pass 'quickstart' "
"for the canned bot.",
),
overwrite_guide: bool = typer.Option(
False,
"--overwrite-guide",
help=f"Overwrite existing guide files ({_AGENTS_FILE}, {_CLAUDE_FILE}, "
f"{_GETTING_STARTED_FILE}). By default existing files are kept.",
),
name: str | None = typer.Option(
None,
"--name",
"-n",
help="Project name (defaults to the target directory name)",
rich_help_panel=_PANEL_SCAFFOLD,
),
bot_type: str | None = typer.Option(
None,
"--bot-type",
"-b",
help="Bot type: 'web' or 'telephony' (inferred from --transport if omitted)",
rich_help_panel=_PANEL_SCAFFOLD,
),
transport: list[str] | None = typer.Option(
None,
"--transport",
"-t",
help="Transport (repeatable, e.g. -t daily -t smallwebrtc)",
rich_help_panel=_PANEL_SCAFFOLD,
),
mode: str | None = typer.Option(
None,
"--mode",
"-m",
help="Pipeline mode: 'cascade' or 'realtime'",
rich_help_panel=_PANEL_SCAFFOLD,
),
stt: str | None = typer.Option(
None, "--stt", help="STT service (cascade mode)", rich_help_panel=_PANEL_SCAFFOLD
),
llm: str | None = typer.Option(
None, "--llm", help="LLM service (cascade mode)", rich_help_panel=_PANEL_SCAFFOLD
),
tts: str | None = typer.Option(
None, "--tts", help="TTS service (cascade mode)", rich_help_panel=_PANEL_SCAFFOLD
),
realtime: str | None = typer.Option(
None,
"--realtime",
help="Realtime service (realtime mode)",
rich_help_panel=_PANEL_SCAFFOLD,
),
video: str | None = typer.Option(
None, "--video", help="Video avatar service", rich_help_panel=_PANEL_SCAFFOLD
),
client_framework: str | None = typer.Option(
None,
"--client-framework",
help="Client framework: 'react', 'vanilla', or 'none'",
rich_help_panel=_PANEL_SCAFFOLD,
),
client_server: str | None = typer.Option(
None,
"--client-server",
help="Client dev server: 'vite' or 'nextjs'",
rich_help_panel=_PANEL_SCAFFOLD,
),
daily_pstn_mode: str | None = typer.Option(
None,
"--daily-pstn-mode",
help="Daily PSTN mode: 'dial-in' or 'dial-out'",
rich_help_panel=_PANEL_SCAFFOLD,
),
twilio_daily_sip_mode: str | None = typer.Option(
None,
"--twilio-daily-sip-mode",
help="Twilio+Daily SIP mode: 'dial-in' or 'dial-out'",
rich_help_panel=_PANEL_SCAFFOLD,
),
config: Path | None = typer.Option(
None,
"--config",
"-c",
help="JSON config file (triggers non-interactive scaffolding)",
rich_help_panel=_PANEL_SCAFFOLD,
),
recording: bool = typer.Option(
False,
"--recording/--no-recording",
help="Enable recording",
rich_help_panel=_PANEL_FEATURES,
),
transcription: bool = typer.Option(
False,
"--transcription/--no-transcription",
help="Enable transcription",
rich_help_panel=_PANEL_FEATURES,
),
video_input: bool = typer.Option(
False,
"--video-input/--no-video-input",
help="Enable video input",
rich_help_panel=_PANEL_FEATURES,
),
video_output: bool = typer.Option(
False,
"--video-output/--no-video-output",
help="Enable video output",
rich_help_panel=_PANEL_FEATURES,
),
deploy_to_cloud: bool = typer.Option(
True,
"--deploy-to-cloud/--no-deploy-to-cloud",
help="Generate cloud deployment files",
rich_help_panel=_PANEL_FEATURES,
),
enable_krisp: bool = typer.Option(
False,
"--enable-krisp/--no-enable-krisp",
help="Enable Krisp noise cancellation",
rich_help_panel=_PANEL_FEATURES,
),
observability: bool = typer.Option(
False,
"--observability/--no-observability",
help="Enable observability",
rich_help_panel=_PANEL_FEATURES,
),
enable_eval: bool = typer.Option(
False,
"--eval/--no-eval",
help="Add an 'eval' transport so the bot is runnable with `-t eval` for "
"behavioral evals (see `pipecat eval`). Off by default.",
rich_help_panel=_PANEL_FEATURES,
),
dry_run: bool = typer.Option(
False, "--dry-run", help="Print resolved scaffold config as JSON without writing files"
),
list_options: bool = typer.Option(
False,
"--list-options",
help="Print available service options as JSON and exit",
callback=list_options_callback,
is_eager=True,
),
):
r"""Initialize a new Pipecat project — and optionally scaffold it.
Writes the coding-agent guide (AGENTS.md, CLAUDE.md). Pass scaffold options (or
``--config``) to also build a runnable bot in place; with none, ``init`` asks how you
want to build.
Examples::
pipecat init # prompt for a directory, then choose how to build
pipecat init my-bot # set up ./my-bot
pipecat init quickstart # canned quickstart bot in ./pipecat-quickstart
pipecat init . # set up the current directory
pipecat init . --bot-type web -t daily -m cascade --stt deepgram_stt --llm openai_llm --tts cartesia_tts
pipecat init my-bot --config project-config.json # scaffold from a config file
pipecat init my-bot --overwrite-guide # refresh existing guide files
pipecat init --list-options # print valid service/transport values as JSON
"""
# `pipecat init quickstart`: scaffold the canned bot in-place, with the coding-agent guide.
if target == "quickstart":
# quickstart is a fixed preset — scaffold flags don't apply. Say so instead of
# silently ignoring them.
if _has_explicit_scaffold_flags(ctx):
console.print(
"[yellow]Note:[/yellow] `quickstart` is a fixed preset; "
"ignoring the scaffold options you passed."
)
return _init_quickstart(overwrite_guide)
scaffold_requested = _scaffold_requested(
config=config,
name=name,
bot_type=bot_type,
transport=transport,
mode=mode,
stt=stt,
llm=llm,
tts=tts,
realtime=realtime,
video=video,
)
if scaffold_requested:
return _scaffold_non_interactive(
ctx,
target=target,
overwrite_guide=overwrite_guide,
dry_run=dry_run,
config=config,
name=name,
bot_type=bot_type,
transport=transport,
mode=mode,
stt=stt,
llm=llm,
tts=tts,
realtime=realtime,
video=video,
client_framework=client_framework,
client_server=client_server,
daily_pstn_mode=daily_pstn_mode,
twilio_daily_sip_mode=twilio_daily_sip_mode,
recording=recording,
transcription=transcription,
video_input=video_input,
video_output=video_output,
deploy_to_cloud=deploy_to_cloud,
enable_krisp=enable_krisp,
observability=observability,
enable_eval=enable_eval,
)
# No scaffold options: write the guide and route to a build method.
# No argument: prompt for the directory. `target` is just a path — `.` sets up the
# current directory, any other value names a folder (created below if it doesn't exist).
if target is None:
target = typer.prompt("Project directory", default="pipecat-bot").strip()
# Validate before any write so a failure never leaves a half-initialized directory.
target_dir = Path(target or ".")
if target_dir.exists() and not target_dir.is_dir():
console.print(f"[red]Error:[/red] {target_dir} exists and is not a directory.")
raise typer.Exit(1)
statuses = _write_agent_guide(target_dir, overwrite_guide)
# A re-run on a guide from an older Pipecat: interactively offer to refresh it now
# (the expected payoff of re-running), otherwise fall back to the printed nudge.
if not overwrite_guide and "kept-stale" in statuses and _is_interactive():
overwrite_guide = _offer_refresh(target_dir)
else:
_print_refresh_summary(statuses)
_route_build_method(target_dir, overwrite_guide)
def _scaffold_non_interactive(
ctx: typer.Context, *, target, overwrite_guide, dry_run, config, **flags
):
"""Write the agent guide and scaffold the project in-place from flags/a config file.
The in-place sibling of the wizard path (:func:`_route_build_method`): same directory,
no ``GETTING_STARTED.md`` (the scaffold's README is the start-here). With no positional
target we scaffold into the current directory rather than prompting, so an automated run
(a coding agent that omits the ``.``) never hangs.
The scaffold config is resolved and validated *before* the guide is written, so an
invalid invocation (or ``--dry-run``) fails without leaving a half-initialized directory.
"""
target_dir = Path(target or ".")
if target_dir.exists() and not target_dir.is_dir():
console.print(f"[red]Error:[/red] {target_dir} exists and is not a directory.")
raise typer.Exit(1)
try:
# Validate first: this exits non-zero on a bad config, or zero on --dry-run, before
# anything is written. Only a fully valid, non-dry-run invocation reaches the writes.
project_config = resolve_scaffold_config(
ctx,
derived_name=target_dir.resolve().name or "pipecat-app",
dry_run=dry_run,
config=config,
**flags,
)
_write_agent_guide(target_dir, overwrite_guide)
generate_scaffold(project_config, dest=target_dir, in_place=True)
except KeyboardInterrupt:
console.print("\n[yellow]Project creation cancelled.[/yellow]")
raise typer.Exit(1)
except typer.Exit:
raise
except FileExistsError as e:
console.print(f"\n[red]Error: {e}[/red]")
raise typer.Exit(1)
except Exception as e:
console.print(f"\n[red]Error creating project: {e}[/red]")
raise typer.Exit(1)