symfonic.agent.cli.main

main

Symfonic agent CLI.

Commands

chat -- Run a single query against the agent. migrate -- Placeholder for future HMS data migration.

Usage

python -m symfonic.agent.cli.main chat "Hello" --tenant-id t1 symfonic chat "Hello" --tenant-id t1 symfonic migrate old://db new://db

chat

chat(
    query: Annotated[
        str, Argument(help="Query to send to the agent.")
    ],
    tenant_id: Annotated[
        str,
        Option(
            "--tenant-id",
            help="Tenant identifier (required).",
        ),
    ],
    dsn: Annotated[
        str | None,
        Option(
            "--dsn",
            help="Postgres DSN for persistent memory.",
        ),
    ] = None,
    model: Annotated[
        str,
        Option(
            "--model",
            help="Model provider name (default: mock).",
        ),
    ] = "mock",
    enable_hms: Annotated[
        bool,
        Option(
            "--enable-hms",
            help="Enable HMS-aware system prompt.",
        ),
    ] = False,
    domain: Annotated[
        str,
        Option(
            "--domain",
            help="Domain template name. Use 'default' for generic agent. Custom domains should be injected programmatically.",
        ),
    ] = "default",
) -> None

Run a single query against the Symfonic agent.

Source code in src/symfonic/agent/cli/main.py

@app.command()
def chat(
    query: Annotated[str, typer.Argument(help="Query to send to the agent.")],
    tenant_id: Annotated[
        str, typer.Option("--tenant-id", help="Tenant identifier (required).")
    ],
    dsn: Annotated[
        str | None, typer.Option("--dsn", help="Postgres DSN for persistent memory.")
    ] = None,
    model: Annotated[
        str, typer.Option("--model", help="Model provider name (default: mock).")
    ] = "mock",
    enable_hms: Annotated[
        bool, typer.Option("--enable-hms", help="Enable HMS-aware system prompt.")
    ] = False,
    domain: Annotated[
        str,
        typer.Option(
            "--domain",
            help="Domain template name. Use 'default' for generic agent. "
            "Custom domains should be injected programmatically.",
        ),
    ] = "default",
) -> None:
    """Run a single query against the Symfonic agent."""
    if dsn:
        typer.echo(
            "Warning: --dsn is not yet wired to Postgres backends. Using in-memory.",
            err=True,
        )

    if model == "mock":
        provider = MockModelProvider(response=f"[mock] Responding to: {query}")
    elif model == "anthropic":
        try:
            from symfonic.core.providers import AnthropicProvider  # pragma: no cover
            provider = AnthropicProvider()  # pragma: no cover
        except ImportError as exc:
            typer.echo(
                "Anthropic provider requires: pip install symfonic-core[anthropic]",
                err=True,
            )
            raise SystemExit(1) from exc
    else:
        typer.echo(f"Unknown model provider: {model}. Use 'mock' or 'anthropic'.", err=True)
        raise SystemExit(1)
    domain_template = DomainTemplate(name=domain)
    config = FrameworkConfig(
        domain=domain_template, enable_hms_prompt=bool(enable_hms),
    )
    agent = SymfonicAgent(model_provider=provider, config=config)
    scope = FrameworkTenantScope(tenant_id=tenant_id)

    if enable_hms:
        print("Thinking (HMS enabled)...", file=sys.stderr)

    async def _run_and_flush() -> AgentResponse:
        response = await agent.run(query, scope=scope)
        await agent.flush_background_tasks()
        return response

    start = time.monotonic()
    response = asyncio.run(_run_and_flush())
    elapsed_ms = (time.monotonic() - start) * 1000

    typer.echo(response.final_response or "(no response)")
    in_tok = len(query) // 3
    out_tok = len(response.final_response or "") // 3
    sys_tok = response.system_prompt_tokens
    total_tok = in_tok + sys_tok + out_tok
    print(
        f"tokens=~{total_tok} (in=~{in_tok} +system=~{sys_tok} out=~{out_tok})  "
        f"memory={response.memory_entries_used}  "
        f"duration_ms={elapsed_ms:.1f}",
        file=sys.stderr,
    )

examples_add

examples_add(
    name: Annotated[
        str,
        Argument(
            help="Example to copy (see 'symfonic examples list')."
        ),
    ],
    dest: Annotated[
        Path,
        Option(
            "--dir",
            "-d",
            help="Destination directory (default: cwd).",
        ),
    ] = Path("."),
    force: Annotated[
        bool,
        Option(
            "--force",
            help="Overwrite an existing target directory.",
        ),
    ] = False,
) -> None

Copy a curated example into <dir>/<name>/ and print how to run it.

Source code in src/symfonic/agent/cli/main.py

@examples_app.command("add")
def examples_add(
    name: Annotated[
        str, typer.Argument(help="Example to copy (see 'symfonic examples list').")
    ],
    dest: Annotated[
        Path,
        typer.Option("--dir", "-d", help="Destination directory (default: cwd)."),
    ] = Path("."),
    force: Annotated[
        bool, typer.Option("--force", help="Overwrite an existing target directory.")
    ] = False,
) -> None:
    """Copy a curated example into ``<dir>/<name>/`` and print how to run it."""
    from symfonic.cli.examples import CURATED, ExamplesError, add_example

    try:
        target = add_example(name, dest, force=force)
    except ExamplesError as exc:
        typer.echo(f"error: {exc}", err=True)
        raise SystemExit(1) from exc

    spec = CURATED[name]
    typer.echo(f"Copied '{name}' -> {target}")
    if spec.requires:
        typer.echo(f"\nRequires: {spec.requires}")
    typer.echo("\nRun it:")
    typer.echo(f"  {spec.run}")

examples_list

examples_list() -> None

List the curated examples available to copy.

Source code in src/symfonic/agent/cli/main.py

@examples_app.command("list")
def examples_list() -> None:
    """List the curated examples available to copy."""
    from symfonic.cli.examples import available_examples

    typer.echo("Available examples (copy with: symfonic examples add <name>):\n")
    width = max(len(s.name) for s in available_examples())
    for spec in available_examples():
        line = f"  {spec.name.ljust(width)}  {spec.summary}"
        if spec.requires:
            line += f"  [requires: {spec.requires}]"
        typer.echo(line)

guide

guide() -> None

Print the agent quick reference (decision tree, imports, footguns).

The reference (AGENTS.md) ships inside the wheel, so this works offline with no repo checkout -- the canonical entry point for an agent or person figuring out how to build with symfonic-core from an installed package.

Source code in src/symfonic/agent/cli/main.py

@app.command()
def guide() -> None:
    """Print the agent quick reference (decision tree, imports, footguns).

    The reference (``AGENTS.md``) ships inside the wheel, so this works offline
    with no repo checkout -- the canonical entry point for an agent or person
    figuring out how to build with symfonic-core from an installed package.
    """
    from importlib.resources import files

    try:
        text = files("symfonic.agent").joinpath("AGENTS.md").read_text(
            encoding="utf-8",
        )
    except (FileNotFoundError, ModuleNotFoundError):  # pragma: no cover
        typer.echo(
            "AGENTS.md is not bundled in this install. See "
            "https://core.symfonic.dev",
            err=True,
        )
        raise SystemExit(1) from None
    typer.echo(text)

init

init(
    project_name: Annotated[
        str | None,
        Argument(
            help="Project name (directory-safe slug). Not required with --list-components or --interactive / -i."
        ),
    ] = None,
    components: Annotated[
        str,
        Option(
            "--components",
            help="Comma-separated list of components to include. Example: --components fastapi,auth,webapp. Default: all components.",
        ),
    ] = _DEFAULT_COMPONENTS,
    list_components: Annotated[
        bool,
        Option(
            "--list-components/--no-list-components",
            help="Print available components and exit.",
            show_default=False,
        ),
    ] = False,
    llm_provider: Annotated[
        str,
        Option(
            "--llm-provider",
            help="Default LLM provider: anthropic, openai, deepseek, kimi, google, ollama, aws, openrouter.",
        ),
    ] = "anthropic",
    output_dir: Annotated[
        Path,
        Option(
            "--output-dir",
            help="Parent directory under which <project_name> will be created.",
        ),
    ] = Path("."),
    interactive: Annotated[
        bool,
        Option(
            "--interactive",
            "-i",
            help="Run interactive scaffolding prompts.",
            show_default=False,
        ),
    ] = False,
    force: Annotated[
        bool,
        Option(
            "--force",
            "-f",
            help="Force scaffolding even if the target directory already exists.",
            show_default=False,
        ),
    ] = False,
) -> None

Scaffold a new symfonic-powered project.

Pass --list-components to print the available component names and exit.

Source code in src/symfonic/agent/cli/main.py

@app.command()
def init(
    project_name: Annotated[
        str | None,
        typer.Argument(
            help="Project name (directory-safe slug). "
            "Not required with --list-components or --interactive / -i.",
        ),
    ] = None,
    components: Annotated[
        str,
        typer.Option(
            "--components",
            help=(
                "Comma-separated list of components to include. "
                "Example: --components fastapi,auth,webapp. "
                "Default: all components."
            ),
        ),
    ] = _DEFAULT_COMPONENTS,
    list_components: Annotated[
        bool,
        typer.Option(
            "--list-components/--no-list-components",
            help="Print available components and exit.",
            show_default=False,
        ),
    ] = False,
    llm_provider: Annotated[
        str,
        typer.Option(
            "--llm-provider",
            help=(
                "Default LLM provider: anthropic, openai, deepseek, kimi, "
                "google, ollama, aws, openrouter."
            ),
        ),
    ] = "anthropic",
    output_dir: Annotated[
        Path,
        typer.Option(
            "--output-dir",
            help="Parent directory under which <project_name> will be created.",
        ),
    ] = Path("."),
    interactive: Annotated[
        bool,
        typer.Option(
            "--interactive",
            "-i",
            help="Run interactive scaffolding prompts.",
            show_default=False,
        ),
    ] = False,
    force: Annotated[
        bool,
        typer.Option(
            "--force",
            "-f",
            help="Force scaffolding even if the target directory already exists.",
            show_default=False,
        ),
    ] = False,
) -> None:
    """Scaffold a new symfonic-powered project.

    Pass --list-components to print the available component names and exit.
    """
    from symfonic.cli.components import ALL_COMPONENTS, COMPONENT_CATALOG

    if list_components:
        typer.echo("Available components:\n")
        for name in ALL_COMPONENTS:
            typer.echo(f"  {name:<20} {COMPONENT_CATALOG[name]}")
        typer.echo(f"\nAll components: {', '.join(ALL_COMPONENTS)}")
        typer.echo(
            f"Default (no --components flag): {_DEFAULT_COMPONENTS}"
            "\n  Example domains are opt-in and mutually exclusive with the generic"
            "\n  starter; add one explicitly, e.g. --components fastapi,webapp,store-starter"
        )
        raise typer.Exit(0)

    # Expand comma-separated component names.
    if interactive:
        if not project_name:
            project_name = typer.prompt("Project name")
            if not project_name or not project_name.strip():
                typer.echo("error: project-name is required.", err=True)
                raise SystemExit(1)

        llm_provider = typer.prompt(
            "Default LLM provider (anthropic, openai, deepseek, kimi, google, "
            "ollama, aws, openrouter)",
            default=llm_provider,
        )

        customize = typer.confirm(
            "Would you like to customize the selected components?",
            default=False,
        )
        if customize:
            selected = []
            for name in ALL_COMPONENTS:
                desc = COMPONENT_CATALOG[name]
                if typer.confirm(f"Include {name} ({desc})?", default=True):
                    selected.append(name)
            if not selected:
                typer.echo("warning: no components selected, using fastapi only.", err=True)
                selected = ["fastapi"]
        else:
            selected = [c.strip() for c in components.split(",") if c.strip()]
    else:
        if not project_name:
            typer.echo(
                "error: project-name is required. "
                "Usage: symfonic init <project-name> [--components ...]",
                err=True,
            )
            raise SystemExit(1)
        selected = [c.strip() for c in components.split(",") if c.strip()]

    # Validate early so the error message comes from the CLI layer.
    unknown = [c for c in selected if c not in set(ALL_COMPONENTS)]
    if unknown:
        typer.echo(
            f"error: unknown component(s): {unknown}. "
            "Run `symfonic init --list-components` to see valid names.",
            err=True,
        )
        raise SystemExit(1)

    try:
        from symfonic.cli.init import ScaffolderError, scaffold
    except ImportError as exc:  # pragma: no cover
        typer.echo(
            "The init command requires jinja2. Install it with: "
            "pip install symfonic-core[init]",
            err=True,
        )
        raise SystemExit(1) from exc

    try:
        scaffold(
            project_name=project_name,
            components=selected,
            llm_provider=llm_provider,
            output_dir=output_dir,
            force=force,
        )
    except ScaffolderError as exc:
        typer.echo(f"error: {exc}", err=True)
        raise SystemExit(1) from exc

migrate

migrate(
    source_uri: Annotated[
        str, Argument(help="Source data URI.")
    ],
    target_uri: Annotated[
        str, Argument(help="Target data URI.")
    ],
) -> None

Migrate HMS data between storage backends (stub).

Source code in src/symfonic/agent/cli/main.py

@app.command()
def migrate(
    source_uri: Annotated[str, typer.Argument(help="Source data URI.")],
    target_uri: Annotated[str, typer.Argument(help="Target data URI.")],
) -> None:
    """Migrate HMS data between storage backends (stub)."""
    typer.echo(f"Migration from {source_uri} to {target_uri} is not yet implemented")
    raise SystemExit(0)