API Reference

Auto-generated from docstrings in the agent_grammar package.

Top-level exports

These three symbols are re-exported from agent_grammar and form the surface you'll touch when annotating a test:

workflow

workflow

workflow(*, name: str, intent: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorate a pytest test function to record its HTTP/boundary steps.

The recorder is stashed on the wrapper as _agent_grammar_recorder so the pytest plugin can collect it after the test passes.

Data flow between steps is not declared by hand; the captured request and response payloads of each step are rendered verbatim (with secrets redacted) so the consuming agent can wire calls together from ground truth.

Source code in src/agent_grammar/testing/decorators.py

def workflow(
    *,
    name: str,
    intent: str,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorate a pytest test function to record its HTTP/boundary steps.

    The recorder is stashed on the wrapper as ``_agent_grammar_recorder`` so the
    pytest plugin can collect it after the test passes.

    Data flow between steps is not declared by hand; the captured request and
    response payloads of each step are rendered verbatim (with secrets
    redacted) so the consuming agent can wire calls together from ground truth.
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        recorder = WorkflowRecorder(name=name, intent=intent)

        if asyncio.iscoroutinefunction(func):

            @functools.wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                token = set_active_recorder(recorder)
                try:
                    result = await func(*args, **kwargs)
                    recorder.mark_complete()
                    return result
                finally:
                    reset_recorder(token)

            setattr(async_wrapper, RECORDER_ATTR, recorder)
            return async_wrapper

        @functools.wraps(func)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            token = set_active_recorder(recorder)
            try:
                result = func(*args, **kwargs)
                recorder.mark_complete()
                return result
            finally:
                reset_recorder(token)

        setattr(sync_wrapper, RECORDER_ATTR, recorder)
        return sync_wrapper

    return decorator

step_boundary

step_boundary

step_boundary(*, domain: str, name: str) -> Iterator[None]

Mark a non-HTTP step (e.g. database query, external service).

Source code in src/agent_grammar/testing/decorators.py

@contextmanager
def step_boundary(*, domain: str, name: str) -> Iterator[None]:
    """Mark a non-HTTP step (e.g. database query, external service)."""
    recorder = get_active_recorder()
    if recorder is not None:
        recorder.add_boundary_step(BoundaryStep(domain=domain, name=name))
    yield

AgentTestClient

AgentTestClient

Bases: TestClient

Drop-in replacement for starlette.testclient.TestClient.

When invoked inside a @workflow-decorated test, each request is captured as an HttpStep and appended to the active recorder.

Data models

HttpStep dataclass

HttpStep(method: str, path: str, request_json: Any | None, status_code: int, response_json: Any | None, domain: str = 'Core Service')

BoundaryStep dataclass

BoundaryStep(domain: str, name: str)

WorkflowRecord dataclass

WorkflowRecord(name: str, slug: str, intent: str, steps: list[Step] = list())

Serving (agent_grammar.serve.fastapi)

GrammarRouter

GrammarRouter

GrammarRouter(filepath: str | Path, *, reload: bool = False)

Return a FastAPI/Starlette router that serves the compiled markdown.

Parameters:

Name	Type	Description	Default
filepath	str | Path	Path to the compiled workflows.md asset.	required
reload	bool	If True, re-read the file on every request (dev mode). If False, cache contents in memory at construction time.	False

Raises:

Type	Description
FileNotFoundError	If filepath does not exist at construction time.
ImportError	If FastAPI is not installed.

Source code in src/agent_grammar/serve/fastapi.py

def GrammarRouter(
    filepath: str | Path,
    *,
    reload: bool = False,
):
    """Return a FastAPI/Starlette router that serves the compiled markdown.

    Args:
        filepath: Path to the compiled workflows.md asset.
        reload: If True, re-read the file on every request (dev mode).
                If False, cache contents in memory at construction time.

    Raises:
        FileNotFoundError: If ``filepath`` does not exist at construction time.
        ImportError: If FastAPI is not installed.
    """
    try:
        from fastapi import APIRouter
    except ImportError as exc:  # pragma: no cover
        raise ImportError(
            "GrammarRouter requires FastAPI. Install with "
            "`pip install agent-grammar[fastapi]`."
        ) from exc

    path = Path(filepath)
    if not path.exists():
        raise FileNotFoundError(
            f"GrammarRouter: workflows file not found at {path!s}. "
            "Did you run pytest to compile it?"
        )

    cached_content: str | None = None
    if not reload:
        cached_content = path.read_text(encoding="utf-8")

    router = APIRouter()

    @router.get("", include_in_schema=False)
    @router.get("/", include_in_schema=False)
    def _serve_workflows() -> PlainTextResponse:
        if reload:
            content = path.read_text(encoding="utf-8")
        else:
            assert cached_content is not None
            content = cached_content
        return PlainTextResponse(
            content=content, media_type=MARKDOWN_MEDIA_TYPE
        )

    return router

AgentTelemetryMiddleware

AgentTelemetryMiddleware

AgentTelemetryMiddleware(app, on_detect: Callable[[str], None] | None = None)

Bases: BaseHTTPMiddleware

Detect requests bearing X-Agent-Grammar-Workflow and report them.

The middleware never raises from the telemetry callback — failures are logged and swallowed so business traffic is unaffected.

Source code in src/agent_grammar/serve/fastapi.py

def __init__(
    self,
    app,
    on_detect: Callable[[str], None] | None = None,
) -> None:
    super().__init__(app)
    self.on_detect = on_detect

CLI (agent_grammar.cli)

The CLI is a Click command group with a single subcommand, export-agent-docs. See Configuration → CLI for flag-by-flag documentation.

export_agent_docs

export_agent_docs(base_url: str, api_version: str, output_dir: str, workflows_path: str | None, platforms: tuple[str, ...]) -> None

Generate platform-specific system prompts for API consumers.

Source code in src/agent_grammar/cli/export.py

@click.command("export-agent-docs")
@click.option(
    "--base-url",
    required=True,
    help="Production base URL of the API (e.g. https://api.example.com).",
)
@click.option(
    "--api-version",
    default="v1",
    show_default=True,
    help="API version namespace (used in the fetch URL and local file path).",
)
@click.option(
    "--output-dir",
    default="./agent-docs",
    show_default=True,
    type=click.Path(file_okay=False, dir_okay=True),
    help="Directory to write the platform-specific system prompts.",
)
@click.option(
    "--workflows-path",
    default=None,
    help=(
        "Local file path where the agent should cache workflows. "
        "Default: .agent/workflows_{version}.md"
    ),
)
@click.option(
    "--platform",
    "platforms",
    multiple=True,
    type=click.Choice(PLATFORMS),
    default=PLATFORMS,
    show_default=True,
    help="Which platform rules to generate. Repeat to select multiple.",
)
def export_agent_docs(
    base_url: str,
    api_version: str,
    output_dir: str,
    workflows_path: str | None,
    platforms: tuple[str, ...],
) -> None:
    """Generate platform-specific system prompts for API consumers."""
    base = base_url.rstrip("/")
    workflows_path = workflows_path or f".agent/workflows_{api_version}.md"
    fetch_url = f"{base}/{api_version}/agent-workflows"

    output = Path(output_dir)
    output.mkdir(parents=True, exist_ok=True)

    env = _build_environment()
    context = {
        "base_url": base,
        "api_version": api_version,
        "workflows_path": workflows_path,
        "fetch_url": fetch_url,
    }

    written: list[Path] = []
    for platform in platforms:
        template = env.get_template(f"{platform}.md.j2")
        rendered = template.render(**context)
        target = output / f"{platform}-rules.md"
        target.write_text(rendered, encoding="utf-8")
        written.append(target)

    click.echo(f"Wrote {len(written)} file(s) to {output}:")
    for path in written:
        click.echo(f"  - {path}")