DEV Community: Andreas Lang

Validators steer, evals grade: picking a model on cost per pass

Andreas Lang — Thu, 13 Aug 2026 13:52:47 +0000

What this post covers

The big changes in this post are:

Add more static validators, like tflint and checkov
Add a structured output type that nudges the agent towards self-reporting
Build an eval set and use pydantic-evals to run it

For any agentic system the cheapest way to get better results is if you can add inexpensive static validators the agent can run itself, and that we re-run on submission. This is particularly true for this agent that produces files on disk. The tools the agent runs and the tools that grade its submission are the same set. We just layer deterministic formatting on top.

How strict the validators should be is a trade-off. checkov ships policies that are far stricter than many projects need, and that context would need to be configured, which we do not want to add right now. So making it slightly less strict is a good way to manage cost (default settings ballooned cost for some models by 100x).

pydantic-evals is the foundation for anything we add later, and eval cases for a real (non-blog example) scenario would need to significantly extend it and add cases that start from an existing project. Currently we have five simple test cases, which does not give us a lot of confidence statistically. Take the results presented with a pinch of salt: they are directionally useful and enough to rule a model out, but they do not predict performance on new and more complex cases.

The work in this post turned up a few gotchas that forced changes to earlier code. With more validation tools the total run size increased, eventually breaching Firehose's record cap (we use it to store audit traces). So we had to add batching for the audit copy. We also cap the model's max_tokens and the run's request_limit, so the message history cannot grow without a ceiling. That does not bound what a tool returns, though MAX_REPORT_CHARS caps the one that mattered, checkov's report. Beyond that we would have to trim or offload the history itself.

From validators to evals

The additional validators we added in this post for the agent answer whether the terraform code is correct, reasonably secure according to static rules, and in line with our lint style. They say nothing about whether it did the task it was assigned. For example if it was tasked to create an AWS S3 bucket and did not, then every one of those checks would still pass.

That is where the evals come in. The structure in pydantic-evals is a small vocabulary. A case is a fixed input plus the metadata that says what should come out of it. The agent runs and results are passed to evaluators. These can be boolean, which gives a pass rate, categorical, which classifies runs and gives a ratio per label, or numeric, which averages into a score. Evaluators can also use OTEL span information meaning even tool calls can be evaluated if tool usage is expected. In addition, token usage and pydantic-ai's genai-prices integration, which we extend with prices it has not shipped yet, give us run cost and cost per pass.

Building an eval harness is the only real way to decide on a model or make an informed change to the setup (like system prompts or tooling). LLM agents are non-deterministic, so without a way to evaluate the results we cannot make any decisions. Collecting stats like token usage, pass rate, and cost is also what lets you run a model to a budget. If your budget is $0.20 per successful run, then you need to know how often you land under it. There is no 100% safety as your evals do not cover everything, which is why you should also do live monitoring for this - the initial premise of our series.

Our eval-case structure is currently basic. We have the inputs (the task, what the agent should do) and some metadata which we use to configure the validators (for example expected terraform resources).

evals/cases.yaml

cases:
  - name: logs-bucket
    inputs: >-
      Create an S3 bucket for storing application logs. Make sure the
      configuration blocks public access.
    metadata:
      expected_resources:
        - type: aws_s3_bucket
        - type: aws_s3_bucket_public_access_block

The result

We did run the eval harness against a few models:

GLM 5.2 (glm5p2), Z.ai's 743B mixture-of-experts model with a 1M-token context, built for long-horizon coding. We reach it through Fireworks.
Mistral Medium (mistral-medium), Mistral's mid-tier generalist, which their own docs position for agentic and coding work.
Haiku 4.5 (haiku), Anthropic's fastest model, and the only Bedrock entry in our registry.
Mistral Large (mistral-large), Mistral's largest general-purpose model.
Codestral (codestral), Mistral's code-completion specialist (not a good choice for the task, which you will see).

Now plotting pass rate against cost per task gives us a decent model selection picture. First of all haiku drops out on cost: it passes as often as mistral-medium and costs more than five times as much per task. glm5p2 is the only model that passed all fifteen runs, but mistral-medium is cheaper per task and its single miss was a run that hit the request limit rather than bad Terraform, so it still comes out ahead on cost per pass. I would expect this to change once we add more complex validation checks and cases, but for now with the eval data we have mistral-medium would be the pick if you can absorb the occasional failure with a retry, and glm5p2 if you want the consistency without one.

See this chart full-size on the original post.

The fuller numbers (n=15: 5 cases x 3 repeats), including the tool-error score:

Model	Pass rate	Errored	Eff $/M tok	Cost / task	Cost / pass	Tool-error
glm5p2	100%	0	$0.68	$0.036	$0.036	0.83
mistral-medium	93%	1	$0.45	$0.028	$0.030	0.66
haiku	93%	1	$1.25	$0.156	$0.167	0.60
mistral-large	87%	1	$2.16	$0.084	$0.097	0.61
codestral	33%	10	$0.33	$0.008	$0.025	0.68

Tool-error is a score between 0 and 1, not a boolean like the evaluators behind the pass rate. It starts at 1 and for each tool call that needs to be retried the score is reduced by 0.1. So you can see that mistral-medium has more tool errors than glm5p2. The causes differ, and some would yield to better instructions. For example in a particular eval case for the log bucket the agent ran terraform init as instructed, then wrote a file adding the first aws provider resource, and then failed on terraform validate with Error: Missing required provider. An instruction covering this edge case would probably have caught it (in real scenarios we should have already aws provider resources in the project). Others are just plain mechanical validations like tflint's terraform "required_version" attribute is required (terraform_required_version). glm5p2 avoids most of these, while mistral-medium got confused enough in one run that it never finished inside the request limit.

What is nice if logfire is configured is the ability to drill into a particular eval case and look at the spans which include the agent conversation and the tool retry errors reducing our score.

interleaving chat spans with tool calls, and three of them flagged as exceptions: terraform\_validate, tflint and
checkov. The detail pane for the selected terraform\_validate span shows a pydantic\_ai ToolRetryError reading
'Error: Missing required provider. This configuration requires provider registry.terraform.io/hashicorp/aws, but
that provider isn't available. You may be able to install it automatically by running: terraform init'." width="800" height="400">

We only have one case where an assertion failed, meaning the model did not fully meet the case's expectations. Most other failures are mechanical, runs that never finished. The 50-request limit is what stops a confused run from spiralling in cost.

See this chart full-size on the original post.

Difficulty is not spread evenly either. queue-with-dlq and sessions-table account for most of the misses, while logs-bucket and lambda-exec-role were passed by every model except codestral, which missed one run of each.

See this chart full-size on the original post.

The code

New to the series? Tooling, AWS access, and project setup are covered in Part 1 (linked above).

The final tree. + is new in post 5, ~ extends a post 4 file, blank carries unchanged. Click any changed or new file to read it; the download below fast-forwards to this state if you want to walk through the post against the finished code.

The following files are new at this checkpoint:

File	What it does
`agent/tools/`	Tools become a package: `filesystem.py` tracks changed files, `validators.py` adds `tflint` and `checkov`
`evals/`	The offline harness: `cases.yaml` and its generated schema, `evaluators.py`, `run.py`. Never ships in the image
`agent/prices.py`	The `glm5p2` price `genai-prices` has not shipped yet. In `agent/`, so a live run is priced like a sweep
`tests/test_evaluators.py`, `tests/fixtures/plan.json`	The plan-comparison logic against a canned plan document
`tests/test_run.py`	The harness points model resolution at the eval registry

These carry forward from post 4 with changes:

File	What changed
`agent/core.py`	`TaskResult` output type and output validator; `execute()` takes an optional workspace, and the run states its own ceilings
`agent/lambda_entry.py`	The response `output` becomes the serialized `TaskResult`
`agent/models.py`	Resolves whichever registry `MODELS_PARAMETER` names, and silences the Mistral SDK's duplicate chat spans
`agent/observability.py`	One Firehose record per span, EMF metrics matched on span scope as well as the `metadata` attribute, and the price registered
`tests/`	`test_core.py`, `test_lambda_entry.py`, `test_tools.py`, `test_observability.py` and `test_validators.py` cover the above

Build and infrastructure:

File	What changed
`infra/models.tf`	A second registry for sweeps, whose Bedrock entry points at its own profile tagged `Purpose=eval`
`infra/lambda.tf`	The Lambda role can read the Fireworks key, so a live run can reach `glm5p2` and not just an eval can
`infra/variables.tf`	`default_model` moves to `mistral-medium`, the model the sweep below argues for
`Dockerfile`	Two more pinned stages: the `tflint` binary, checksum-verified, and a `checkov` venv
`pyproject.toml`	The dev group gains `pydantic-evals` and `typer`

Two rows in those tables have a story behind them. The audit copy from post 2 shipped a whole trace as one Firehose record, which a long validation run outgrew, so it now ships one record per span, and the run states its own ceilings so no single span can outgrow the cap either.

The other is cost. An uncapped checkov report is not paid for once: it joins the message history and is charged again on every later turn, so validators.py caps what a failed check hands back to the model.

terraform-pr-agent/
    agent/
      tools/
      + __init__.py
      + filesystem.py
      + validators.py
      __init__.py
    ~ core.py
      env.py
    ~ lambda_entry.py
      memory.py
    ~ models.py
    ~ observability.py
    + prices.py
      runs.py
      ssm.py
    evals/
    + __init__.py
    + cases_schema.json
    + cases.yaml
    + evaluators.py
    + run.py
    infra/
      placeholder/
        Dockerfile
        handler.py
      alerts.tf
      audit-bucket.tf
      bedrock.tf
      cloudwatch.tf
      ecr.tf
      firehose.tf
      iam.tf
      kms.tf
    ~ lambda.tf
      logfire.tf
      main.tf
    ~ models.tf
      runs-bucket.tf
    ~ variables.tf
    scripts/
      build-lambda.sh
      chat.py
      queries.sql
      traces.sql
    tests/
      fixtures/
      + plan.json
    + clean.tf
      conftest.py
    ~ test_core.py
      test_env.py
    + test_evaluators.py
    ~ test_lambda_entry.py
    + test_observability.py
    + test_run.py
    ~ test_tools.py
    + test_validators.py
    .dockerignore
    .envrc
    .envrc.local
    .gitignore
    AGENTS.md
  ~ Dockerfile
  ~ pyproject.toml
    README.md

Browse these files interactively on the original post.

Fast-forward to the final code of this post:

mkdir -p ~/projects
cd ~/projects
curl -fsSL https://andreaslang.dev/terraform-pr-agent/terraform-pr-agent-05.tar.gz | tar xz

More validation tools

We extend the validation suite well beyond Post 4's terraform_validate. It is nice to tell the agent "make sure the S3 bucket you create follows security best practices", but it is better to use checkov to check that the bucket is encrypted and has a lifecycle policy that deletes old versions and tflint to check that nothing uses deprecated syntax or known bad practices.

We do not want to give the agent raw access to the shell and all features of terraform, so we create a wrapper class that standardises how each shell command runs and how its output or failure is reported.

agent/tools/validators.py

class CommandResult(BaseModel):
    success: bool
    stdout: str
    stderr: str

    def format_error_for_agent(self) -> str:
        """The failure as the model sees it, capped. Callers keep the full output."""
        if not self.success:
            return f"failed:\n{_capped(self.stdout)}\n{_capped(self.stderr)}"
        raise RuntimeError("CommandResult.format_error_for_agent() called on success")


class Command(BaseModel):
    name: str
    commands: list[str]

    @property
    def installs_providers(self) -> bool:
        return "init" in self.commands

    def run(self, path: Path) -> CommandResult:
        if self.installs_providers:
            with _INSTALL_LOCK:
                return self._run(path)
        return self._run(path)

    def _run(self, path: Path) -> CommandResult:
        # Inside the lock, so the span measures the command and not the wait.
        with track_memory(self.name):
            result = subprocess.run(self.commands, cwd=path, capture_output=True, text=True)
            return CommandResult(
                success=result.returncode == 0, stdout=result.stdout, stderr=result.stderr
            )

    def run_root(self, ctx: RunContext[WorkspaceDeps]) -> CommandResult:
        return self.run(ctx.deps.root)

    def run_in_tool(self, path: Path) -> str:
        result = self.run(path)
        if not result.success:
            raise ModelRetry(f"{self.name} {result.format_error_for_agent()}")
        return f"OK: {self.name} passed."

The first runs with the new tools went badly, and logfire showed it straight away. Token usage was rising quickly and I could see a lot of failed tool calls for checkov. Failed here means the checks did not pass, not that the tool crashed. A quick dive into the tool parameters showed that checkov was returning a very long report where:

All checks, even the successful ones, were listed, filling the agent's context and confusing it
The actual failed tests were extremely strict, stricter than you would expect in a normal setup

For example KMS over AES256 is a good rule, but irrelevant outside a compliance setting. Similarly, every S3 bucket is told it needs its own access log bucket. We disable a short list of rules, but only when the workspace ships no .checkov.yaml of its own.

So the fix? Add --quiet --compact to checkov, so we do not fill the agent's context with irrelevant information. I would rather read the short version as a human too. Together with excluding a few rules, the results improved immediately. We went from timing out after 300s and 2M tokens input to 94s and less than 20k tokens input. Significant cost difference!

A failed check is not paid for once, though. Its output joins the message history and is billed again on every later turn, so we also cap what a failure hands back:

agent/tools/validators.py

# Every failure report becomes a tool result the model carries for the rest of
# the run, so an uncapped one is paid for on every later turn and lands in the
# audit record. checkov's report is the offender; its head carries the
# failures that matter, and the agent re-runs the check anyway.
MAX_REPORT_CHARS = 8_000


def _capped(text: str) -> str:
    if len(text) <= MAX_REPORT_CHARS:
        return text
    return f"{text[:MAX_REPORT_CHARS]}\n[{len(text) - MAX_REPORT_CHARS} more characters cut]"

agent/tools/validators.py

TERRAFORM_INIT = Command(
    name="terraform_init",
    commands=["terraform", "init", "-backend=false", "-input=false", "-no-color"],
)

TERRAFORM_VALIDATE = Command(
    name="terraform_validate",
    commands=["terraform", "validate", "-no-color"],
)

TERRAFORM_FMT = Command(
    name="terraform_fmt",
    commands=["terraform", "fmt", "-recursive"],
)

TFLINT = Command(
    name="tflint",
    commands=["tflint", "--format", "compact"],
)

# Checks skipped by default: cost or architecture posture decisions, not
# security baselines, and routinely disabled in real projects. Genuine
# baselines (public access blocks, encryption at rest, IAM wildcards) stay on.
_CHECKOV_DEFAULT_SKIPS = [
    "CKV_AWS_18",  # S3 access logging on every bucket
    "CKV_AWS_144",  # S3 cross-region replication
    "CKV_AWS_145",  # S3 must use KMS; SSE-S3 (CKV_AWS_19) still enforced
    "CKV2_AWS_61",  # S3 lifecycle configuration on every bucket
    "CKV2_AWS_62",  # S3 event notifications on every bucket
    "CKV_AWS_50",  # Lambda X-Ray tracing
    "CKV_AWS_115",  # Lambda reserved concurrency
    "CKV_AWS_116",  # Lambda dead-letter queue
    "CKV_AWS_117",  # Lambda attached to a VPC
    "CKV_AWS_272",  # Lambda code signing
    "CKV_AWS_338",  # CloudWatch log retention of at least a year
]


def _checkov_command(root: Path) -> Command:
    """Default skips apply only when the workspace brings no config of its own:
    checkov auto-discovers .checkov.yaml in the scanned directory, and a project
    that states its policy wins over ours.
    """
    commands = ["checkov", "-d", ".", "--quiet", "--compact"]
    if not any((root / name).exists() for name in (".checkov.yaml", ".checkov.yml")):
        commands += ["--skip-check", ",".join(_CHECKOV_DEFAULT_SKIPS)]
    return Command(name="checkov", commands=commands)

Wrapping them as tools is then trivial. The command objects never mutate their state, so holding them as module-level singletons carries no concurrency risk.

agent/tools/validators.py

def terraform_init(ctx: RunContext[WorkspaceDeps]) -> str:
    """Run ``terraform init`` in the workspace.

    Required once before the first ``terraform_validate`` and again after
    provider or module requirements change.
    """
    return TERRAFORM_INIT.run_in_tool(ctx.deps.root)


def terraform_validate(ctx: RunContext[WorkspaceDeps]) -> str:
    """Run ``terraform validate`` in the workspace and return its output."""
    return TERRAFORM_VALIDATE.run_in_tool(ctx.deps.root)


def tflint(ctx: RunContext[WorkspaceDeps]) -> str:
    return TFLINT.run_in_tool(ctx.deps.root)


def checkov(ctx: RunContext[WorkspaceDeps]) -> str:
    return _checkov_command(ctx.deps.root).run_in_tool(ctx.deps.root)

validate_workspace is different: we hook it up as an output validator. In the previous post we ran this check ourselves and passed the messages back into a new run with the retry prompt. Then I discovered pydantic-ai covers this out of the box, which let us delete most of that code.

agent/tools/validators.py

def validate_workspace(path: Path) -> CommandResult:
    # Normalize formatting before gating; a fmt failure (unparsable HCL) is
    # ignored here because terraform validate reports it better one step later.
    TERRAFORM_FMT.run(path)
    for command in [TERRAFORM_VALIDATE, TFLINT, _checkov_command(path)]:
        result = command.run(path)
        if not result.success:
            break
    return result

We extend the system prompt so the agent reaches for the new validators:

Run tflint and checkov and clear all findings before reporting done.

`TaskResult`: structured self-report

Making the agent declare what it did, knowing the claim gets checked, changes its behaviour more than I expected. So in this post we also added a structured output type:

agent/core.py

class TaskResult(BaseModel):
    """The agent's structured self-report on a finished run.

    Each required field pushes the agent to consider that dimension of its
    work; the tool-call spans in the trace remain the ground truth that
    exposes any embellishment. Because this is the agent's output type,
    ending a run now requires calling the final_result output tool, so a
    text-only reply can no longer end a run silently.
    """

    model_config = ConfigDict(frozen=True)

    summary: str
    """One line on what was done."""
    solution_description: str
    """How the problem was solved, including architectural choices."""
    validations_run: list[str]
    """Which validation tools were invoked during the run."""
    issues_addressed: list[str]
    """Security or correctness problems identified and fixed."""
    known_limitations: list[str]
    """What was not handled; surfaces for human review."""
    ready_for_review: bool
    """The agent's self-assessment that the workspace is PR-ready."""

Reporting done is not the same as being done, so the same validate_workspace gate runs as an output validator: the agent only escapes the loop when the checks it is told to run actually pass.

agent/core.py

_RETRY_PROMPT = (
    "terraform validate still reports errors after you reported done. "
    "Fix them and validate again.\n\n{output}"
)


@agent.output_validator
def _validate_final_workspace(ctx: RunContext[WorkspaceDeps], output: TaskResult) -> TaskResult:
    # The deliverable is the workspace, not this object: a run that changed
    # nothing produced nothing, however plausible its self-report reads.
    if not ctx.deps.files_changed:
        raise ModelRetry(
            "You reported done but made no changes to the workspace. "
            "Use the file tools to implement the request, validate, then report done again."
        )
    validation_result = validate_workspace(ctx.deps.root)
    if not validation_result.success:
        raise ModelRetry(_RETRY_PROMPT.format(output=validation_result.format_error_for_agent()))
    return output

To be clear this does not entirely prevent the agent from making things up and filling things in it did not do, but it does push more runs the right way, because each output field forces the agent to account for that dimension.

In the eval runs I compared different models with and without the structured output. In particular weaker ones like Haiku (vs glm5p2) did better if they were forced to justify themselves in the structured output. glm5p2 showed no difference. For Haiku the structured output runs were cleaner with tool calls 27.0 down to 22.4, validator calls 13.0 down to 10.8, validate-pass 93% up to 100%, errored runs 6.7% down to 0%.

The output type itself is enforced, the agent has to call a tool with this enforced schema to end the run. We also have a max limit of turns and errors to avoid infinite unsuccessful runs.

This structured schema is also what will let us deterministically create PRs. This is what we set out to do in the first place after all.

The eval harness: the validators are the conditions

pydantic_evals is built around a few concepts:

Datasets: contain cases for evaluation and which Evaluators to apply - the central abstraction validation runs against
Cases: inputs, expected outputs, and metadata
Evaluators: Logic to evaluate which can be based on expected outputs and metadata

It offers a few built-in evaluators, but the more complex checks we need, actually inspecting the code the agent produced, required building our own.

WorkspaceValidates - uses the existing validate_workspace output validator to check the workspace is valid. Technically a slight double validation as the agent cannot submit the result without this.
PlanMatchesGraph - runs terraform plan -out and terraform show -json to compare the workspace to the expected graph (coming from metadata). Configuring the AWS provider authenticates against STS, so the plan would need real credentials. It runs on a copy of the workspace with a *_override.tf that supplies mock ones, which keeps the sweep offline and leaves the graded workspace untouched.
SelfReportAccurate - the agent has to say which checks it did run in its TaskResult, we validate this against actual tool calls by going through recorded spans.
ToolErrorMetricEvaluator - slightly different from the others: here we give a numeric score from 0.0 to 1.0 to indicate how many tool calls failed. Every time a tool call fails 0.1 is subtracted from the score until it reaches 0.0.

One slight practical hiccup was when the agent decided to add variables without a default, which made it difficult to create a plan. Therefore we instruct the agent to default every variable it declares and fail the evaluation if it does not.

WorkspaceValidates is the one place the policy tools and the harness meet, in eight lines: it calls the same validate_workspace() the agent had to satisfy at runtime, so a case clears that check for the same reason the agent was allowed to stop. The other three evaluators ask questions the validators cannot answer.

evals/evaluators.py

@dataclass
class WorkspaceValidates(Evaluator[TASK_INPUTS, EvalOutput, CaseMetadata]):
    """fmt + terraform validate + tflint + checkov, verbatim from the agent."""

    def evaluate(
        self, ctx: EvaluatorContext[TASK_INPUTS, EvalOutput, CaseMetadata]
    ) -> EvaluationReason:
        result = validate_workspace(ctx.output.workspace)
        if result.success:
            return EvaluationReason(value=True)
        return EvaluationReason(value=False, reason=result.format_error_for_agent())

Take SelfReportAccurate, one of the more interesting evaluators here: you can see that the context passed to the evaluator contains a ctx.span_tree object with functions to find spans we need for evals. The downside is that it does not carry non-genai attributes, which is a problem if you are looking for error information that is recorded in an OTEL-specific field. Not a problem here, but was an issue for ToolErrorMetricEvaluator as the errors were harder to identify consistently.

evals/evaluators.py

@dataclass
class SelfReportAccurate(Evaluator[TASK_INPUTS, EvalOutput, CaseMetadata]):
    """The TaskResult agrees with what the case knows.

    The tool-call spans in the trace stay the ground truth. This checks only the
    self-report claims that a case can verify cheaply.
    """

    def evaluate(
        self, ctx: EvaluatorContext[TASK_INPUTS, EvalOutput, CaseMetadata]
    ) -> dict[str, EvaluatorOutput]:
        report = ctx.output.result
        validations = " ".join(report.validations_run).lower()
        tool_spans = ctx.span_tree.find(
            predicate=SpanQuery(
                name_contains="execute_tool", has_attribute_keys=["gen_ai.tool.name"]
            )
        )
        actual_tools = {span.attributes.get("gen_ai.tool.name") for span in tool_spans}
        return {
            "ready_for_review": EvaluationReason(
                value=report.ready_for_review,
                reason="Agent should mark ready for review.",
            ),
            "reported_linters": EvaluationReason(
                value=(
                    "tflint" in validations
                    and "checkov" in validations
                    and "tflint" in actual_tools
                    and "checkov" in actual_tools
                ),
                reason=(
                    f"Mismatch in validations: {validations} and {actual_tools}, "
                    "should both contain tflint and checkov."
                ),
            ),
        }

Running the sweep

Now you can run all eval cases across the full suite of models. The command below will run 5 models and repeat each model three times. LLMs are non-deterministic, so three repeats give enough spread to tell a real failure from a bad roll.

Sweep the registry

uv run python -m evals.run \
  --model glm5p2 --model haiku --model mistral-large \
  --model mistral-medium --model codestral --repeat 3

You get one experiment per model, each with its own pass rate.

assertion pass rate, and a Recent experiments table with one row per model run (haiku, codestral, mistral-medium,
mistral-large, glm5p2) showing pass rates from 70% to 100%." width="1595" height="1227">

Once you open a particular eval run you can see more detailed results, including our assertion/boolean validators and the tool error metric score. Further down are operational metrics like cost and average task duration.

63 of 63 assertions passed, no task errors, 171.51s average task duration. Evaluator analysis lists PlanMatchesGraph,
ready\_for\_review, reported\_linters, reported\_planted\_issue and WorkspaceValidates at 100%, with tool\_error\_score at
0.64 as a distribution. Operational metrics show 50,510 input tokens, 1,728 output tokens and $0.02366 cost." width="1595" height="1227">

You can also interact with the logfire MCP to create ad-hoc summary tables, or, if you will run it repeatedly, have a model write a script against the API. That is where the numbers at the top of this post come from.

End state

In this post we gave the agent a stricter definition of correct Terraform. The same fmt, validate, tflint and checkov steer it while it works and gate what it submits. None of that says whether it did the task it was asked to do, which is the question the eval harness answers.

We implemented an eval harness that lets us weigh different models against each other and allows us to make a choice based on numbers instead of gut-feel.

In the next post we will integrate with github and ship the PR flow, that means human in the loop via github and deciding how we manage state across the whole PR lifecycle.

File tools and validate: the agent's feedback loop

Andreas Lang — Thu, 25 Jun 2026 10:51:04 +0000

What this post covers

This post gives the agent real work. In previous posts we just gave it toy work to make sure the infrastructure, monitoring, and audit logs work. Now we build a harness for the agent to write Terraform and validate it.

To do this we give the agent a few tools:

Sandboxed File System Tools:
- list_files: list files in a folder.
- read_file: read a file.
- write_file: write a file.
- edit_file: edit a file (search and replace based).
- delete_file: delete a file.
Terraform Tools:
- terraform_init: initialize a Terraform workspace.
- terraform_validate: validate a Terraform workspace.

The sandboxed file system tools themselves are simple; most of the work is returning good error messages to the model and making sure it does not breach the boundaries of the sandbox tmp folder. The terraform tools on the other hand are more complex, and they have consequences for the Lambda setup. They need to actually execute the terraform binary. For this post that meant changing the Lambda from a Python to a Docker based Lambda, which did require changing the build process and infrastructure setup.

Typical agent flow looks like this:

agent invokes file write tool, terraform init, and terraform validate." width="749" height="297">

Notice the new AWS Lambda span, which was not there in the previous post. This is because I wanted to extend instrumentation given that now there is more happening than just the agent run. We do also re-lock the terraform dependency lock file for different architectures after the agent run and I wanted to see timings and memory usage for the external terraform process. Because we are using AWS Lambda, memory is a constraint and cost factor, made worse by the fact that if you are (like me) running this in a new sandbox AWS account you are capped at 3008 MB of memory. Increasing that is a slow process on basic support.

The system prompt in this iteration is not optimized (we'll do that in a later post). We just tell the agent what tools it has and that it needs to run terraform-init before the first validate. list_files/read_file is currently not that important as we do not hook up GitHub/existing terraform code in this post. It will be more important when the agent needs to assess how to integrate its code into an existing project and to follow that project's best practice guidelines when doing so.

System Prompt

You are the terraform-pr-agent. You operate on a Terraform workspace
through file tools (list_files, read_file, write_file, edit_file, delete_file)
and two terraform tools. Use the file tools to explore, write, and edit.
Run terraform_init before your first validate and again whenever you add
or change provider or module requirements. Call terraform_validate after
you write or change files to confirm the workspace still parses; treat its
output as feedback and edit until it is clean.

When you add the AWS provider, give it a version constraint such as "~> 6.0"
rather than leaving it unconstrained. terraform init records the exact resolved
version and checksums in .terraform.lock.hcl, which travels with the workspace
and is the reproducibility record, so the constraint does not need to be an exact
pin. Pin an exact version only when the user asks for one. Example:

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 6.0"
    }
  }
}

Similarly, the user prompt is basic, with not much thought put into it. It is passed in the invoke event payload.

User Prompt

Set up a new terraform project, creating a best practice s3 bucket.

We store the end result in an S3 bucket. With no GitHub integration yet, that is a simple way to inspect the files the agent produced and validate the outcome.

Architecture

Post 2 wrapped the agent in a Lambda and stood up the dual-sink span pipeline behind it; post 3 made the model a runtime choice (an SSM model registry, a Mistral provider beside Bedrock, and EMF metrics feeding one CloudWatch dashboard). Post 4 keeps all of that as is. The Lambda itself changes from a zip package to a container image so we can ship the terraform CLI alongside the Python runtime; a new ECR repository holds the image, and terraform seeds it with a placeholder so the function can be created before the first real push, the container twin of post 2's placeholder zip. Everything else around the function (IAM, Bedrock, the model registry, Firehose, S3) carries forward unchanged. The rest of the post is software: a workspace under /tmp, file tools, the terraform_validate tool, and a caller-side retry.

See this diagram full-size on the original post.

The code

New to the series? Tooling, AWS access, and project setup are covered in Part 1 (linked above).

The final tree. + is new in post 4, ~ extends a post 3 file, blank carries unchanged. Click any changed or new file to read it; the download below fast-forwards to this state if you want to walk through the post against the finished code. We also split the old handler.py in two: core.py holds the agent and an execute(prompt, model) that knows nothing about Lambda, and lambda_entry.py is the Lambda boundary that parses the event and calls it. It was getting out of hand as this post's functionality piled on.

terraform-pr-agent/
    agent/
      __init__.py
    + core.py
    + env.py
    + lambda_entry.py
    + memory.py
    + models.py
    + observability.py
    + runs.py
    + ssm.py
    + tools.py
    infra/
      placeholder/
      + Dockerfile
      + handler.py
      alerts.tf
      audit-bucket.tf
      bedrock.tf
    ~ cloudwatch.tf
    + ecr.tf
      firehose.tf
      iam.tf
      kms.tf
    ~ lambda.tf
      logfire.tf
      main.tf
      models.tf
    + runs-bucket.tf
    ~ variables.tf
    scripts/
    ~ build-lambda.sh
      chat.py
      queries.sql
    ~ traces.sql
    tests/
    ~ conftest.py
    + test_core.py
    + test_env.py
    + test_lambda_entry.py
    + test_tools.py
  + .dockerignore
    .envrc
    .envrc.local
    .gitignore
    AGENTS.md
  + Dockerfile
  ~ pyproject.toml
  + README.md

Browse these files interactively on the original post.

Fast-forward to the final code of this post:

mkdir -p ~/projects
cd ~/projects
curl -fsSL https://andreaslang.dev/terraform-pr-agent/terraform-pr-agent-04.tar.gz | tar xz

From zip to container image

In previous posts we used a normal Python Lambda, but for this post we moved to a Docker based Lambda to avoid the Lambda 250 MB size limit (unzipped). We could still have managed in this post with terraform 108 MB + site-packages 56 MB = 164 MB, but the container ceiling is 10 GB rather than 250 MB, which buys real headroom as the image grows. The container also hands us the whole image to control, system packages and filesystem layout, not just Python deps layered onto the AWS runtime. As a bonus it is a standard Docker image, so moving to something like ECS Fargate later is straightforward.

Looking at the Docker image you will also notice we use a multi-stage build, where we have build layers of the image in multiple stages and then copy them into the final image. This is an easy way of keeping build tools out of the final image and therefore reducing the size of the image. Our final image is 739 MB total, 169 MB above the shared 570 MB base. That sounds huge for people trying to build small images, but the base image we use is the AWS Lambda Python base image, which is heavily cached across the Lambda stack. So in practice we only pull the final 169 MB layer, which is far more reasonable.

Multi-stage build layout

builder stage        copied into the final image
-------------        ---------------------------
terraform       -->  terraform binary
python          -->  site-packages (/var/task)
insights        -->  Lambda Insights extension
build context   -->  agent/ source

base image: public.ecr.aws/lambda/python:3.13
left in the builders, never shipped: unzip, dnf cache, uv, gpg keyring, rpm metadata

We will not go through all the stages of the image, but here are the Python dependencies as an example. We use the AWS Lambda Python base image, install uv by copying the layer from yet another Docker image, copy pyproject.toml/uv.lock, export to a requirements.txt and then run uv pip to install the dependencies.

Dockerfile

# Same uv flow as the zip build this replaces (see the uv AWS Lambda
# guide); the dependency set installs straight into the task root, where
# the final stage picks it up.
FROM public.ecr.aws/lambda/python:3.13 AS python
COPY --from=ghcr.io/astral-sh/uv:0.11.21 /uv /usr/local/bin/uv
WORKDIR /opt/build
COPY pyproject.toml uv.lock ./
RUN uv export --frozen --no-dev --no-editable -o requirements.txt \
    && uv pip install \
        --no-installer-metadata \
        --no-compile-bytecode \
        --target "${LAMBDA_TASK_ROOT}" \
        -r requirements.txt

The Python stage does not ship. The final stage below just assembles the image by copying all the previous layers into it. Here you see we use:

The Terraform binary
The Lambda Insights extension (extensions and cloudwatch)
The Python dependencies
Our agent code

Initially I considered baking the Terraform AWS provider into the image, but managing it properly turned out to be difficult without constraining the use case further than it already is. So I took the simpler route: download the providers every time the agent calls terraform init.

Dockerfile

# The shipped image: only the artifacts the function uses at runtime
# cross over from the builder stages.
FROM public.ecr.aws/lambda/python:3.13
COPY --from=terraform /usr/local/bin/terraform /usr/local/bin/terraform
COPY --from=insights /opt/extensions /opt/extensions
COPY --from=insights /opt/cloudwatch /opt/cloudwatch
COPY --from=python ${LAMBDA_TASK_ROOT} ${LAMBDA_TASK_ROOT}
COPY agent ${LAMBDA_TASK_ROOT}/agent
WORKDIR ${LAMBDA_TASK_ROOT}

The CMD attribute for a Docker based Lambda needs to be the Python handler. It points at the handler function in the agent.lambda_entry module. While we are at it we also make terraform non-interactive and set the home folder to the only writable folder in a Lambda (/tmp).

Dockerfile

# terraform needs a writable HOME for incidental state; /tmp is the only
# writable path at runtime. CMD replaces the zip package's handler attribute
# and points at lambda_entry, the instrumented entry point, not the bare
# handler module.
ENV HOME=/tmp \
    TF_IN_AUTOMATION=true \
    TF_INPUT=false
CMD ["agent.lambda_entry.handler"]

ECR and the placeholder image

A consequence of switching to a Docker based Lambda is that we now need a placeholder image to avoid the terraform apply needing a three step process (first creating the ECR repo, pushing the image, then creating the Lambda). With the placeholder in place, a single apply can create the function, and scripts/build-lambda.sh pushes the real image afterward. It keeps the terraform apply self-contained, with the code ship as a separate step.

infra/ecr.tf

# Container twin of the zip flow's archive_file placeholder: the function
# resource needs a pullable image at create time, so terraform seeds a
# minimal one. Create-only (input never changes), so scripts/build-lambda.sh
# owns every push after this.
resource "terraform_data" "placeholder_image" {
  input = aws_ecr_repository.agent.repository_url

  # Needs docker and the aws cli on the machine running apply; both are
  # already prerequisites for the series.
  provisioner "local-exec" {
    command = <<-EOT
      aws ecr get-login-password --region ${data.aws_region.current.region} |
        docker login --username AWS --password-stdin ${split("/", aws_ecr_repository.agent.repository_url)[0]}
      docker buildx build --platform linux/arm64 --provenance=false \
        -t ${aws_ecr_repository.agent.repository_url}:placeholder \
        --push ${path.module}/placeholder
    EOT
  }
}

For the Lambda itself, package_type = "Image" and image_uri switch it to a Docker based Lambda. We also tell terraform to ignore changes to image_uri, because otherwise a re-apply would reset the function back to the placeholder image. Instead we want our agent Docker build to have control over the image URI. Further, we tweak timeout, memory size, and ephemeral storage to match the heavier resource needs now that terraform runs inside the function. Lambda layers, which we used in a previous post, are removed entirely; they do not work with a Docker based Lambda.

infra/lambda.tf

resource "aws_lambda_function" "agent" {
  function_name = "terraform-pr-agent"
  role          = aws_iam_role.lambda.arn
  architectures = ["arm64"]

  # The handler entry point comes from the image's CMD; the runtime,
  # handler, and layers attributes only apply to zip packages.
  package_type = "Image"
  image_uri    = "${aws_ecr_repository.agent.repository_url}:placeholder"

  # Max Memory Used overstates what this function needs. It runs to ~2 GB on a
  # heavy run, but the track_memory spans (see agent/memory.py) show the real,
  # non-reclaimable demand stays under ~1 GB: ~315 MB resident for the Python
  # runtime plus a transient ~420 MB while terraform validate loads the
  # provider schema. The rest is reclaimable page cache from the ~800 MB
  # provider download and re-lock unpacks doing file IO on /tmp, which the
  # cgroup-based billed figure counts but the kernel drops under pressure, so
  # it is not OOM risk. Memory is therefore not the binding constraint here.
  # 3008 is set for the vCPU it buys, not the RAM: above 1769 MB Lambda gives a
  # full core (3008 is ~1.7), which speeds the run. Drop it toward ~1769 if
  # latency matters less than cost; do not raise it for memory headroom.
  memory_size = 3008

  # The per-run tool budget is the runaway guard; the timeout only has to
  # accommodate several model turns with init + validate rounds in between.
  timeout = 300

  # Two things land in /tmp: terraform init downloads the AWS provider
  # (~800 MB) into the workspace, and the post-run re-lock unpacks the
  # provider for three platforms (another ~2 GB) to write a portable lock
  # file. 4 GB covers both with headroom; the default 512 MB would not.
  ephemeral_storage {
    size = 4096
  }

  tracing_config {
    mode = "Active"
  }

  environment {
    variables = merge(
      {
        MODELS_PARAMETER         = aws_ssm_parameter.models.name
        DEFAULT_MODEL            = var.default_model
        METRICS_NAMESPACE        = local.metrics_namespace
        FIREHOSE_DELIVERY_STREAM = aws_kinesis_firehose_delivery_stream.audit.name
        RUNS_BUCKET              = aws_s3_bucket.runs.bucket
      },
      local.logfire_token_wired ? {
        LOGFIRE_TOKEN_PARAMETER = local.logfire_token_parameter_name
      } : {},
      local.mistral_key_wired ? {
        MISTRAL_API_KEY_PARAMETER = aws_ssm_parameter.mistral_api_key[0].name
      } : {},
    )
  }

  # Code ships out of band: scripts/build-lambda.sh pushes a new image and
  # calls update-function-code, so terraform must not flip the function
  # back to the placeholder on the next apply.
  lifecycle {
    ignore_changes = [image_uri]
  }

  depends_on = [
    aws_iam_role_policy_attachment.lambda_basic_execution,
    aws_iam_role_policy_attachment.lambda_insights,
    aws_iam_role_policy.lambda_permissions,
    terraform_data.placeholder_image,
  ]
}

Watching cold starts

One common downside of a Docker Lambda is that cold starts are slower than normal Lambdas. We do not particularly care about cold starts, but we do want to measure them and make sure they do not get out of hand due to dependency bloat. The AWS Lambda Insights extension is a good way to measure cold starts, but because we cannot use layers, we need to add it to the image. The code looks scary, and I wish AWS thought about the end user the way Astral/uv do, but it is taken straight from the AWS docs.

Dockerfile

# Container images cannot attach layers, so the Lambda Insights extension
# is baked into the image: pinned version, detached GPG signature checked
# against the key fingerprint published in the Lambda Insights docs, so a
# tampered rpm fails the build.
FROM public.ecr.aws/lambda/python:3.13 AS insights
ARG INSIGHTS_VERSION=1.0.660.0
ARG INSIGHTS_BASE_URL=https://lambda-insights-extension-arm64.s3-ap-northeast-1.amazonaws.com
# The downloaded key is checked against the fingerprint from the docs
# before anything trusts it; gpg runs with --batch/--no-tty/--no-autostart
# because the base image ships no gpg-agent.
RUN curl -fsSLO ${INSIGHTS_BASE_URL}/amazon_linux/lambda-insights-extension-arm64.${INSIGHTS_VERSION}.rpm \
    && curl -fsSLO ${INSIGHTS_BASE_URL}/amazon_linux/lambda-insights-extension-arm64.${INSIGHTS_VERSION}.rpm.sig \
    && curl -fsSLO ${INSIGHTS_BASE_URL}/lambda-insights-extension.gpg \
    && gpg --batch --no-tty --show-keys --with-colons lambda-insights-extension.gpg \
        | grep -q '^fpr:::::::::E0AFFA11FFF35BD7349EE222479C97A1848ABDC8:' \
    && gpg --batch --no-tty --no-autostart --import lambda-insights-extension.gpg \
    && gpg --batch --no-tty --no-autostart --verify \
        lambda-insights-extension-arm64.${INSIGHTS_VERSION}.rpm.sig \
        lambda-insights-extension-arm64.${INSIGHTS_VERSION}.rpm \
    && rpm -U lambda-insights-extension-arm64.${INSIGHTS_VERSION}.rpm \
    && rm -f lambda-insights-extension-arm64.${INSIGHTS_VERSION}.rpm* lambda-insights-extension.gpg

The extension needs one IAM grant to write to its /aws/lambda-insights log group, a single managed-policy attachment that sits in the file browser above.

Below you see how we wire the metrics into the dashboard. We take the Lambda Insights metrics and plot:

avg duration and max duration in ms
function memory used
/tmp storage space used

infra/cloudwatch.tf

      {
        type   = "text"
        x      = 0
        y      = 0
        width  = 24
        height = 2
        properties = {
          markdown = "## Lambda\nContainer-image function health: invocations and errors, end to end duration, cold start init duration (Lambda Insights emits `init_duration` only on a cold start), and the memory and `/tmp` footprint behind the `memory_size` and `ephemeral_storage` sizing."
        }
      },
      {
        type   = "metric"
        x      = 0
        y      = 2
        width  = 12
        height = 6
        properties = {
          title  = "Lambda invocations and errors"
          region = local.cloudwatch_region
          view   = "timeSeries"
          stat   = "Sum"
          period = 60
          metrics = [
            ["AWS/Lambda", "Invocations", "FunctionName", local.lambda_name, { label = "${local.lambda_name} / invocations" }],
            [".", "Errors", ".", ".", { label = "${local.lambda_name} / errors" }],
            [".", "Throttles", ".", ".", { label = "${local.lambda_name} / throttles" }],
          ]
        }
      },
      {
        type   = "metric"
        x      = 12
        y      = 2
        width  = 12
        height = 6
        properties = {
          title  = "Lambda duration (ms)"
          region = local.cloudwatch_region
          view   = "timeSeries"
          period = 60
          metrics = [
            ["AWS/Lambda", "Duration", "FunctionName", local.lambda_name, { label = "${local.lambda_name} / avg", stat = "Average" }],
            [".", ".", ".", ".", { label = "${local.lambda_name} / p99", stat = "p99" }],
          ]
        }
      },
      {
        type   = "metric"
        x      = 0
        y      = 8
        width  = 12
        height = 6
        properties = {
          title  = "Cold start init duration (ms)"
          region = local.cloudwatch_region
          view   = "timeSeries"
          period = 60
          metrics = [
            # Insights reports init_duration only when an init phase happened,
            # so points appear only on cold starts.
            ["LambdaInsights", "init_duration", "function_name", local.lambda_name, { label = "${local.lambda_name} / init avg (ms)", stat = "Average" }],
            [".", ".", ".", ".", { label = "${local.lambda_name} / init max (ms)", stat = "Maximum" }],
          ]
        }
      },
      {
        type   = "metric"
        x      = 12
        y      = 8
        width  = 12
        height = 6
        properties = {
          title  = "Memory used (MB)"
          region = local.cloudwatch_region
          view   = "timeSeries"
          period = 60
          metrics = [
            # used_memory_max is the cgroup figure (Max Memory Used), which
            # counts the reclaimable /tmp page cache and so reads ~2 GB while
            # real demand is under 1 GB. Backs the memory_size comment in
            # lambda.tf.
            ["LambdaInsights", "used_memory_max", "function_name", local.lambda_name, { label = "${local.lambda_name} / memory max (MB)", stat = "Maximum" }],
          ]
        }
      },
      {
        type   = "metric"
        x      = 0
        y      = 14
        width  = 12
        height = 6
        properties = {
          title  = "/tmp used (bytes)"
          region = local.cloudwatch_region
          view   = "timeSeries"
          period = 60
          metrics = [
            # tmp_used tracks the ~800 MB provider download into /tmp, behind
            # the 4 GB ephemeral_storage sizing in lambda.tf.
            ["LambdaInsights", "tmp_used", "function_name", local.lambda_name, { label = "${local.lambda_name} / tmp used (B)", stat = "Maximum" }],
          ]
        }
      },

In CloudWatch the dashboard looks like this. Cold starts sit around 3s, acceptable for our use case, and the memory and /tmp metrics stay within the configured limits.

starts, memory and /tmp) above a Model section (tokens, invocations and errors, latency, cache tokens)." width="1721" height="1173">

Workspace setup

The agent needs some context for the run. Pydantic-ai is itself stateless. The agent only has conversation history if the previous messages are passed to it as a message_history argument. We talked about this back in post 1. Here we also need to give the agent context about the workspace it is working in so that the root/project folder can be validated. Tool calls should usually just use a relative path to it, but we need to make sure everything ends up in this tmp folder, so we can pick up the end result of the agent's work. You also see files_read there; we use it to stop the agent editing files it has not read.

agent/tools.py

class WorkspaceDeps(BaseModel):
    """Per-run dependencies threaded through ``RunContext``.

    ``root`` is the directory the agent is allowed to read, write, and
    validate inside. Tools resolve every path relative to it and reject
    anything that escapes the root, so the agent cannot reach outside
    the workspace via ``..`` or absolute paths.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)

    root: Path
    files_read: set[Path] = Field(default_factory=set)

Each time the LLM invokes any of our tools with a path we validate it against the workspace root. Luckily Python has the pathlib module that makes this easy.

agent/tools.py

def _resolve_absolute_path(ctx: RunContext[WorkspaceDeps], path: str):
    root = ctx.deps.root.resolve()
    absolute_path = (root / Path(path)).resolve()
    if not absolute_path.is_relative_to(root):
        raise ModelRetry(
            f"Path {path} must be relative to the workspace root, "
            f"it cannot be absolute or walk up the directory tree."
        )
    return absolute_path

File tools

We have list_files(path), read_file(path), write_file(path, contents), edit_file(path, old_string, new_string) and delete_file(path). We won't go through all of them, but show you the edit file tool. It uses _resolve_absolute_path(ctx, path) to resolve the path (always absolute in the end), which will also verify that the path is inside the workspace root. Then it validates additional constraints:

The file exists
The agent has previously read or created the file
old_string (string to replace) is in the file contents
old_string only exists once in the file contents (prevents accidental replacements)

One decision worth calling out: we raise ModelRetry instead of returning a description of the mistake the agent makes. The difference is that ModelRetry counts against the tool retry budget of the agent, while a string description telling the agent what to do next does not. It avoids long chains of corrections costing too many tokens. If that happens we would rather fail and raise an alert in a production environment, so that an SRE or the product team can investigate and fix the cause of the agent's confusion (better system prompt or tool descriptions). During initial validation it also allows you to weed out models that are not as good with tool calls, which makes them unsuitable for your task.

agent/tools.py

def edit_file(
    ctx: RunContext[WorkspaceDeps],
    path: str,
    old_string: str,
    new_string: str,
) -> None:
    """Replace ``old_string`` with ``new_string`` in the file at ``path``."""
    file = _resolve_absolute_file(ctx, path)
    if not file.exists():
        raise ModelRetry(f"File {path} does not exist.")
    if file not in ctx.deps.files_read:
        raise ModelRetry(f"File {path} was not read. If you want to edit it, then read it first.")
    with file.open("r+") as f:
        contents = f.read()
        if old_string not in contents:
            raise ModelRetry(f"String {old_string} not found in file {path}.")
        if contents.count(old_string) > 1:
            raise ModelRetry(f"String {old_string} found more than once in file {path}.")
        f.seek(0)
        f.write(contents.replace(old_string, new_string))
        f.truncate()

The validate tool: agent-driven, any time

In this blog post we only offer terraform_validate() and terraform_init() to the agent to interact and validate terraform code. Terraform validate for example has a very simple flow:

Use a sub-process to run terraform validate
Validate the result:
- If the exit code is 0 (success) return OK message
- If the exit code is 1 (failure) ModelRetry error with stdout and stderr

agent/tools.py

def validate_workspace(root: Path) -> tuple[bool, str]:
    """Run ``terraform validate`` in ``root``; return (passed, combined output).

    Called by the ``terraform_validate`` tool and by core.py's caller-side
    retry, which re-checks the workspace after the run.
    """
    with track_memory("terraform_validate"):
        result = subprocess.run(
            ["terraform", "validate", "-no-color"],
            cwd=root,
            capture_output=True,
            text=True,
        )
    return result.returncode == 0, f"{result.stdout}{result.stderr}"


def terraform_validate(ctx: RunContext[WorkspaceDeps]) -> str:
    """Run ``terraform validate`` in the workspace and return its output."""
    ok, output = validate_workspace(ctx.deps.root)
    if ok:
        return "OK: terraform validate passed."
    raise ModelRetry(f"terraform validate failed:\n{output}")

A little bonus is the track_memory decorator, which creates a new span and attaches memory information to it. So we can dive into the agent run in Logfire and see how much memory it used.

spans from the track\_memory decorator, each tagged with mem.used\_bytes, mem.cache\_bytes, and mem.child\_max\_rss\_bytes attributes." width="800" height="425">

This is the Agent definition, one instance reused across invocations. The error budget for ModelRetry can be configured in the Agent setup.

agent/core.py

# A soft reproducibility nudge following the standard Terraform pattern: a
# version constraint in the config, the exact version and checksums in the lock
# file. The tool-call spans in the trace are the ground truth for what the agent
# actually wrote.
_PROVIDER_PIN_RULE = (
    "When you add the AWS provider, give it a version constraint such as "
    '"~> 6.0" rather than leaving it unconstrained. terraform init records '
    "the exact resolved version and checksums in .terraform.lock.hcl, "
    "which travels with the workspace and is the reproducibility record, "
    "so the constraint does not need to be an exact pin. Pin an exact "
    "version only when the user asks for one. Example:\n"
    "\n"
    "terraform {\n"
    "  required_providers {\n"
    "    aws = {\n"
    '      source  = "hashicorp/aws"\n'
    '      version = "~> 6.0"\n'
    "    }\n"
    "  }\n"
    "}"
)

SYSTEM_PROMPT = (
    "You are the terraform-pr-agent. You operate on a Terraform workspace "
    "through file tools (list_files, read_file, write_file, edit_file, "
    "delete_file) and two terraform tools. Use the file tools to explore, "
    "write, and edit. Run terraform_init before your first validate and "
    "again whenever you add or change provider or module requirements. "
    "Call terraform_validate after you write or change files to confirm "
    "the workspace still parses; treat its output as feedback and edit "
    "until it is clean.\n\n" + _PROVIDER_PIN_RULE
)

# One Agent instance is reused across invocations: the tools reach the workspace
# through RunContext.deps, so each run_sync scopes them to a fresh WorkspaceDeps.
# The model carries no default; it is built from the registry at INVOKE and
# passed per run, so switching DEFAULT_MODEL needs no code change.
agent = Agent(
    deps_type=WorkspaceDeps,
    system_prompt=SYSTEM_PROMPT,
    tools=[
        list_files,
        read_file,
        write_file,
        edit_file,
        delete_file,
        terraform_init,
        terraform_validate,
    ],
    # Tools raise ModelRetry on failure; pydantic-ai ends the run once one tool
    # fails more than `retries` times in a row (a success resets the count). The
    # default of 1 would end the run on the second straight failing validate,
    # which is a normal part of the write-validate-edit loop, so the budget is
    # raised well past anything a converging run produces. The per-run turn cap
    # stays the runaway guard.
    retries=10,
)

Caller-side retry

Working with LLMs reminds me of the German saying:

Vertrauen ist gut, Kontrolle ist besser (Trust is good, verifying is better)

Here the agent can in principle ignore every instruction and report success while nothing actually works. Hence, we add some deterministic checks in the end and feed message_history back to the agent with a new prompt if they fail. At this stage this is really just making sure terraform init does not fail.

agent/core.py

            # The agent can report done while terraform validate still fails.
            # Re-validate ourselves and feed any error back as a follow-up turn,
            # reusing the run id and message history so each retry is an
            # invoke_agent span under the one invocation trace. Give up after the
            # budget and raise so the failure is honest rather than a clean run
            # over a broken workspace.
            ok, output = validate_workspace(root)
            attempts = 0
            while not ok and attempts < _MAX_VALIDATE_RETRIES:
                attempts += 1
                result = agent.run_sync(
                    _RETRY_PROMPT.format(output=output),
                    deps=deps,
                    conversation_id=run_id,
                    model=built,
                    message_history=result.all_messages(),
                    metadata={"model": model_name},
                )
                ok, output = validate_workspace(root)
            if not ok:
                raise ValidateDidNotConverge(output)

Parking the output in S3

In this iteration of the code we do not yet integrate GitHub, so to be able to see the code that the agent produced, we need to store it somewhere. This code stores it in an S3 bucket so we can inspect it later. We skip the .terraform folder, which is just local init scratch.

agent/runs.py

def _persist_run(run_id: str, workspace: Path, status: str, error: str | None = None) -> None:
    """Park the workspace and a minimal result marker under runs/<run_id>/.

    result.json carries only what the audit trace does not: prompt, output, and
    messages already live in the trace under the same conversation id, so copying
    them here would create a second source of truth.
    """
    bucket = require_env("RUNS_BUCKET")
    s3 = boto3.client("s3")
    for file in _workspace_files(workspace):
        key = f"runs/{run_id}/workspace/{file.relative_to(workspace)}"
        s3.put_object(Bucket=bucket, Key=key, Body=file.read_bytes())
    result = {"status": status} | ({"error": error} if error else {})
    s3.put_object(
        Bucket=bucket,
        Key=f"runs/{run_id}/result.json",
        Body=json.dumps(result).encode(),
    )


def _workspace_files(workspace: Path) -> Iterator[Path]:
    """Every file except .terraform/, which is init scratch plus the provider
    downloaded into /tmp, gigabytes of noise per run. The top-level
    .terraform.lock.hcl is the reproducibility record and stays.
    """
    for path in sorted(workspace.rglob("*")):
        if ".terraform" in path.relative_to(workspace).parts:
            continue
        if path.is_file():
            yield path

Tracing the whole invocation

The Lambda entry point stays thin: it requires a prompt, hands off to core.execute, and shapes the response. Its other job is the INIT wiring, configure telemetry then wrap the handler, which has to run at import. Keeping the execution logic separate from anything Lambda-related also means we could add other entry points later, deploying this somewhere else (e.g. ECS Fargate).

agent/lambda_entry.py

"""The Lambda boundary: parse the event, run the agent, shape the response.

This module also owns the INIT wiring. The container CMD targets
agent.lambda_entry.handler, so a unit test that imports agent.core never
configures logfire or registers the Firehose audit processor. Everything
Lambda-specific lives here, off the core, which is why no runtime-detection
check is needed to keep it out of tests.
"""

from typing import NotRequired, TypedDict

import logfire

from agent import observability
from agent.core import execute


class HandlerEvent(TypedDict):
    prompt: str
    model: NotRequired[str]


class HandlerResponse(TypedDict):
    status: str
    run_id: str
    model: str
    output: str


def handler(event: HandlerEvent, context: object) -> HandlerResponse:
    """Lambda entry point: require a prompt, run the agent, wrap the result.

    ``prompt`` is required; an event without one is a caller error and fails
    fast rather than running a default. ``model`` overrides DEFAULT_MODEL when
    given. A run that does not converge raises, so the Lambda reports 5xx and
    the workspace ships under status error for debugging.
    """
    prompt = event.get("prompt")
    if not prompt:
        raise ValueError("event missing required 'prompt'")
    result = execute(prompt, event.get("model"))
    return {
        "status": "ok",
        "run_id": result.run_id,
        "model": result.model,
        "output": result.output,
    }


def bootstrap() -> None:
    """Stand up telemetry, then attach the Lambda runtime adapter.

    configure() first so the tracer provider exists when the handler is wrapped.
    instrument_aws_lambda wraps the target named by _HANDLER
    (agent.lambda_entry.handler) in place, so each invocation becomes one trace.
    """
    observability.configure()
    logfire.instrument_aws_lambda(handler)


bootstrap()

Changing the span hierarchy also means the example query we showed in a previous post for reading the audit data in S3 needs to change. We already changed the audit processor to ship if span.parent is None or span.parent.is_remote, now in the query we look for the agent invocations in particular by using gen_ai.operation.name = 'invoke_agent'. This works with or without the Lambda OTEL data being written.

scripts/traces.sql

roots AS (
    -- One row per agent run: pydantic-ai's invoke_agent span, identified by
    -- the GenAI operation rather than by being the parentless span.
    -- instrument_aws_lambda now roots each trace at the SpanKind.SERVER
    -- invocation span, so the agent run is a child of it, not the trace root.
    -- (A caller-side retry would put several invoke_agent spans under one
    -- invocation; the trace_id join below would then cross them, so that
    -- case wants each chat tied to its enclosing run instead.)
    SELECT * FROM spans
    WHERE list_filter(attributes, x -> x.key = 'gen_ai.operation.name')[1]
        .value.stringValue = 'invoke_agent'
),

What validate catches and what it doesn't

Currently we only run terraform validate, which checks that the HCL and provider config are correct. It does not catch bad naming for Terraform variables, security issues, or misconfigurations that only show up during apply. In the following post we remedy some of this by giving the agent more tools to validate against, and enforcing extra checks such as security baselines.

End state

An agent that takes an English request and leaves a workspace in a validated state. File tools, validate tool, retry chain, and a structured self-report are all in place for posts 4-6 to build on.

Next: Conventions and policy: more tools, same feedback loop

Model portability: swapping Bedrock for the Mistral API

Andreas Lang — Tue, 16 Jun 2026 11:32:05 +0000

New to the series? Tooling, AWS access, and project setup are covered in Part 1 (linked above).

What this post covers

Recently the US government decided to put export controls in place for Anthropic Mythos and Fable models. See here for details. While this is only for the recently released Fable/Mythos models, it did get me thinking about the increasing risk of reliance on US only foundation models. While I am obviously aware that this post is still running on AWS, I wanted to at least make a move to a European foundational model.

Admittedly, there is not a grand deal of choice and it also meant moving away from AWS Bedrock. Bedrock does have a few Mistral models, but regions are extremely inflexible and the specific one I wanted to use (Mistral Large 3) was not available in the EU at all (the model card says so, but it is not). Losing Bedrock also meant losing direct integration with CloudWatch, but luckily the decision to go with OTLP for audit meant I already had the code hooked up to extract these metrics out of the trace. That in combination with EMF (Embedded Metrics Format) meant I could easily send these as custom metrics to CloudWatch without a great deal of code changes.

Originally I had planned to only add the ability to switch between the models later when we get to evaluation, but with the recent events I changed the order, so the new code does still support Haiku via Bedrock, but added also the ability to use Mistral models via Mistral's API.

The final tree. + is new in post 3, ~ extends a post 2 file, blank carries unchanged. Click any changed or new file to read it; the download below fast-forwards to this state.

terraform-pr-agent/
    agent/
      __init__.py
    ~ handler.py
    infra/
    ~ alerts.tf
      audit-bucket.tf
      bedrock.tf
    ~ cloudwatch.tf
      firehose.tf
      iam.tf
      kms.tf
    ~ lambda.tf
      logfire.tf
      main.tf
    + models.tf
    ~ variables.tf
    scripts/
      build-lambda.sh
      chat.py
      queries.sql
      traces.sql
    tests/
    + conftest.py
    + test_handler.py
    .envrc
  ~ .envrc.local
    .gitignore
    AGENTS.md
  ~ pyproject.toml

Browse these files interactively on the original post.

Fast-forward to the final code of this post:

mkdir -p ~/projects
cd ~/projects
curl -fsSL https://andreaslang.dev/terraform-pr-agent/terraform-pr-agent-03.tar.gz | tar xz

To use Mistral models, you will need to create an API key and configure it in your .envrc.local file. Sign up here and create an API key here. For this post's usage the free tier is fine, but you may as well load 10 Euros on it and switch to the "Scale" plan of the API. Otherwise you will very quickly receive 429 errors.

Architecture

Post 2 ran a single Bedrock model behind the Lambda and shipped spans to Logfire and the S3 audit copy. Post 3 keeps that intact and turns the model into a runtime choice: Terraform renders a model registry into SSM Parameter Store, the handler builds the pydantic-ai model on first invoke by reading that registry, and a Mistral API entry sits alongside the Bedrock one (with the Mistral key fetched from SSM the same way as the Logfire token). Metrics move to EMF, so a Bedrock model and a Mistral-API model land in the same CloudWatch namespace and one dashboard covers both.

See this diagram full-size on the original post.

The model registry

To support both models I am passing a simple config via AWS SSM Parameter Store into the Lambda. It defines provider model id and if on bedrock inference profile to be used.

infra/models.tf

# The model registry: Terraform owns it, renders it to JSON, and parks it in
# an SSM String parameter the handler reads at startup. Each entry names a
# provider and a model id; Bedrock entries also carry the inference-profile
# ARN. DEFAULT_MODEL (set on the Lambda) selects the active one, so switching
# the agent's model is a parameter change, not a code change.
locals {
  metrics_namespace = "TerraformPrAgent/Models"

  models = {
    haiku = {
      provider              = "bedrock"
      model_id              = local.bedrock_model_id
      inference_profile_arn = aws_bedrock_inference_profile.agent.arn
    }
    "mistral-large" = {
      provider = "mistral"
      model_id = "mistral-large-latest"
    }
    "devstral-small" = {
      provider = "mistral"
      model_id = "devstral-small-2507"
    }
  }

  mistral_key_wired = var.mistral_api_key != ""
}

resource "aws_ssm_parameter" "models" {
  name        = "/terraform-pr-agent/models"
  description = "Model registry for the terraform-pr-agent Lambda (provider + model id per entry)."
  type        = "String"
  value       = jsonencode(local.models)
}

In addition we need a Mistral API key wired and retrieved the same way as the Logfire key via SSM Parameter Store (encrypted).

infra/models.tf

# The Mistral API key, SecureString, fetched by the handler through the same
# Parameters and Secrets extension path as the Logfire token. Only created
# when TF_VAR_mistral_api_key is set, mirroring the Logfire token wiring; with
# it unset the Mistral providers are simply unreachable and a Bedrock default
# still works.
resource "aws_ssm_parameter" "mistral_api_key" {
  count = local.mistral_key_wired ? 1 : 0

  name        = "/terraform-pr-agent/mistral-api-key"
  description = "Mistral API key. Consumed by the terraform-pr-agent Lambda."
  type        = "SecureString"
  value       = var.mistral_api_key
}

Building the model at invoke time

Now that we support Bedrock and Mistral models, we just need to create the right pydantic-ai model object with the matching configuration. The handler has also been modified so the model to be used can be provided via the event payload. The default is Mistral Large 3 if nothing is provided.

agent/handler.py

@cache
def _build_model(name: str) -> Model:
    """Build the pydantic-ai model registered under ``name``.

    The registry lives in an SSM String parameter, so this runs on the first
    INVOKE (the extension is not ready during INIT) and is memoised per model
    name for warm invocations. Bedrock models authenticate via the Lambda
    role; Mistral models read an API key from a SecureString parameter,
    fetched the same way as the Logfire token.
    """
    registry = json.loads(_fetch_ssm_parameter(os.environ["MODELS_PARAMETER"]))
    config = registry[name]
    provider = config["provider"]
    if provider == "bedrock":
        return BedrockConverseModel(
            config["model_id"],
            settings={"bedrock_inference_profile": config["inference_profile_arn"]},
        )
    if provider == "mistral":
        key_param = os.environ.get("MISTRAL_API_KEY_PARAMETER")
        if not key_param:
            raise RuntimeError(
                f"model {name!r} uses the Mistral API, but MISTRAL_API_KEY_PARAMETER "
                "is not set. Set MISTRAL_API_KEY and re-apply so the key is wired, or "
                "select a Bedrock model via DEFAULT_MODEL or the event's model field."
            )
        return MistralModel(
            config["model_id"],
            provider=MistralProvider(
                api_key=_fetch_ssm_parameter(key_param),
                http_client=_retrying_http_client(),
            ),
        )
    raise ValueError(f"unknown provider {provider!r} for model {name!r}")

Provider-agnostic metrics with EMF

To avoid having one model via the inference profile and the Mistral models via a different mechanism, we switch all models to use EMF logged metrics, so we can build a clean dashboard (check it in the code you can download above).

agent/handler.py

def _emit_emf(spans: Sequence[ReadableSpan]) -> None:
    """Emit one EMF metric line for the trace, read off the root span.

    pydantic-ai records gen_ai.usage.* on the root agent span as the run total
    (the sum of its child chat spans), so a single read is the correct total,
    not a sum across every span. The model dimension is the registry key the
    handler passed as run metadata; pydantic-ai serialises that to the root
    span's `metadata` attribute (even on a failed run), so it is read back here
    rather than carried in module state. That key is exactly what the dashboard
    iterates, so a Bedrock run and a Mistral run share one set of widgets.
    Logging the _aws envelope to stdout is enough; CloudWatch Logs extracts the
    metrics from the structured line.
    """
    root = next((span for span in spans if span.parent is None), None)
    if root is None:
        return
    attributes = root.attributes or {}
    model = json.loads(attributes.get("metadata", "{}")).get("model", "unknown")
    errored = root.status.status_code is StatusCode.ERROR
    record = {
        "_aws": {
            "Timestamp": root.end_time // 1_000_000,
            "CloudWatchMetrics": [
                {
                    "Namespace": os.environ["METRICS_NAMESPACE"],
                    "Dimensions": [["Model"]],
                    "Metrics": [
                        {"Name": "InputTokens", "Unit": "Count"},
                        {"Name": "OutputTokens", "Unit": "Count"},
                        {"Name": "CacheReadTokens", "Unit": "Count"},
                        {"Name": "CacheWriteTokens", "Unit": "Count"},
                        {"Name": "Latency", "Unit": "Milliseconds"},
                        {"Name": "Invocations", "Unit": "Count"},
                        {"Name": "Errors", "Unit": "Count"},
                    ],
                }
            ],
        },
        "Model": model,
        "InputTokens": attributes.get("gen_ai.usage.input_tokens", 0),
        "OutputTokens": attributes.get("gen_ai.usage.output_tokens", 0),
        # pydantic-ai sets these only when non-zero, so default to 0. Providers
        # without prompt caching (e.g. the Mistral API) simply never report them.
        "CacheReadTokens": attributes.get("gen_ai.usage.cache_read.input_tokens", 0),
        "CacheWriteTokens": attributes.get("gen_ai.usage.cache_creation.input_tokens", 0),
        "Latency": (root.end_time - root.start_time) / 1_000_000,
        "Invocations": 1,
        "Errors": 1 if errored else 0,
    }
    log.info("trace_metrics", **record)


def _on_trace_complete(spans: Sequence[ReadableSpan]) -> None:
    """Ship the audit copy, then emit metrics: one hook, two sinks."""
    _ship_trace(spans)
    _emit_emf(spans)

You might also wonder about log.info("trace_metrics", **record) and how this logs in the right format for EMF. Well, the answer is I sneaked in structlog. It is an amazing Python logging library that has all the things and ease of use the standard logging library misses.

agent/handler.py

# JSON logs to stdout, which CloudWatch Logs ingests as-is. The same stream also
# carries the EMF metric envelope (see _emit_emf), so one structured sink covers
# both application logs and metrics. Logging has no extension dependency, so it
# is configured at import rather than on the first INVOKE.
structlog.configure(
    processors=[
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.EventRenamer("message"),
        structlog.processors.JSONRenderer(),
    ],
    logger_factory=structlog.PrintLoggerFactory(),
    cache_logger_on_first_use=True,
)
log = structlog.get_logger()

End State

Ease of switching between models and EMF logging/monitoring configured and the ability to run a (good) European foundation model 🇪🇺!

Coming next: workspace and small toolkit for the agent to get to work.

Audit trail via OTLP: every agent run as a trace

Andreas Lang — Tue, 16 Jun 2026 11:31:36 +0000

New to the series? Tooling, AWS access, and project setup are covered in Part 1 (linked above).

Optional: Logfire for the live trace UI

Logfire is the live visualizer we screenshot in the querying section. It is optional: the span processor we set up below only enables the Logfire leg when a write token is present. You can finish the post without it and query the audit copy directly via Athena; the rest of the setup works either way.

If you want it, sign up at logfire.pydantic.dev (GitHub SSO works for the personal tier), create a project named terraform-pr-agent, then open Project settings -> Write tokens -> New write token. There is nothing finer-grained to pick: Logfire write tokens are project-scoped write keys by design. They cannot read telemetry, list other projects, delete the project, or impersonate users, so the blast radius if the token leaks is "someone can write garbage spans into this one project." Name the token terraform-pr-agent-collector so it is obvious what it belongs to, and copy the value immediately (the UI shows it once).

Then add it to .envrc.local and run direnv reload:

.envrc.local

export LOGFIRE_TOKEN="pylf_v1_..."

If you skip this section, leave LOGFIRE_TOKEN unset. The collector config below uses a Terraform conditional on the variable being non-empty, so the otlphttp exporter and its IAM permission only materialise when the token is there.

What this post covers

Every agent run becomes an OpenTelemetry trace. We wire pydantic-ai's instrumentation to an in-process span processor in the Lambda itself, fanning OTLP out to two destinations: an S3 bucket with Object Lock in compliance mode for the immutable audit copy, and Logfire as the live visualizer for debugging the same trace. Logfire is not the system of record. S3 is.

The final tree. + is new in post 2, ~ extends a post 1 file, blank carries unchanged. Click any changed or new file to read it; the download below fast-forwards to this state if you want to walk through the post against the finished code.

terraform-pr-agent/
    agent/
    + __init__.py
    + handler.py
    infra/
      alerts.tf
    + audit-bucket.tf
      bedrock.tf
      cloudwatch.tf
    + firehose.tf
      iam.tf
    + kms.tf
    + lambda.tf
    + logfire.tf
      main.tf
    ~ variables.tf
    scripts/
    + build-lambda.sh
      chat.py
    + queries.sql
    + traces.sql
    .envrc
  ~ .envrc.local
    .gitignore
    AGENTS.md
  + pyproject.toml

Browse these files interactively on the original post.

Fast-forward to the final code of this post:

mkdir -p ~/projects
cd ~/projects
curl -fsSL https://andreaslang.dev/terraform-pr-agent/terraform-pr-agent-02.tar.gz | tar xz

Architecture

Post 1 was a local script calling Bedrock through an assumed role, with CloudWatch metrics as the only observation surface. Post 2 wraps that call in a Lambda, adds the dual-sink span pipeline (logfire to the live UI, an in-memory buffer to Firehose for the audit copy), and stands up the S3-side query layer on top.

See this diagram full-size on the original post.

OTLP schema for audit

We do not design a schema ourselves. The OpenTelemetry GenAI semantic conventions already specify what an agent run looks like as a trace, and pydantic-ai's Logfire instrumentation emits that shape for us. Reusing the convention means the same trace renders in Logfire, Jaeger, Tempo, or any OTLP-compatible backend, and attributes like gen_ai.usage.input_tokens mean the same thing everywhere.

ℹ️ Note

The GenAI semantic conventions are still marked Development, but DataDog and Logfire already implement them - Logfire even versions its support so in-development spec changes don't break existing traces. That matters here: if the spec shifts, we can still assert that every attribute the audit trail depends on is present and roll out changes safely.

A trace is a tree of spans sharing one trace_id. Each span has a parent, a start and end time, a status, and a bag of attributes. Per the OTel spec, span names follow {operation} {target}: the agent run becomes a root invoke_agent span, each LLM call becomes a chat span named after the model, and each tool becomes an execute_tool span. Pydantic-ai emits these names when you opt into the spec-compliant instrumentation with logfire.instrument_pydantic_ai(version=5). Rendered as a waterfall in a typical trace UI, one agent run looks like this:

invoke_agent terraform_pr_agent  ████████████████  4.1s
  chat claude-haiku-4-5          █████             0.9s
  execute_tool write_file             █            0.2s
  chat claude-haiku-4-5                ███         0.6s
  execute_tool terraform_validate         ██       0.3s

Each chat span carries the GenAI attributes that matter: gen_ai.request.model, gen_ai.usage.input_tokens and output_tokens, gen_ai.input.messages and gen_ai.output.messages (when content recording is on), and gen_ai.response.finish_reasons. Each execute_tool span carries the tool name, arguments, and result. Concretely:

chat claude-haiku-4-5
  gen_ai.system                  aws.bedrock
  gen_ai.operation.name          chat
  gen_ai.request.model           claude-haiku-4-5
  gen_ai.usage.input_tokens      1247
  gen_ai.usage.output_tokens     312
  gen_ai.response.finish_reasons ["stop"]
  gen_ai.input.messages          [{role: user, parts: [...]}]
  gen_ai.output.messages         [{role: assistant, parts: [...]}]

execute_tool write_file
  gen_ai.operation.name          execute_tool
  gen_ai.tool.name               write_file
  gen_ai.tool.call.id            toolu_01ABC...
  gen_ai.tool.call.arguments     {path: main.tf, content: ...}
  gen_ai.tool.call.result        {ok: true, bytes_written: 412}

That is the free debugger. When a run misbehaves you have a complete record of what the agent saw, what it called, what the call returned, and what it decided next. The pydantic-ai Logfire integration docs list everything emitted by default.

Infra

Variables

The full variables.tf with post 1's two carried through and three new ones for the audit pipeline (retention horizon, Parameters-and-Secrets layer ARN, and the Logfire write token):

infra/variables.tf

variable "alert_email" {
  description = "Email address subscribed to the agent alerts SNS topic. Set via TF_VAR_alert_email."
  type        = string
}

variable "daily_token_alarm_threshold" {
  description = "Daily combined input + output token threshold. Crossing it sends an email via SNS."
  type        = number
  default     = 1000000
}

variable "audit_retention_days" {
  type        = number
  description = "Object Lock default retention in days. Tutorial default is 7 so the bucket is easy to clean up; production audit horizons are typically years (e.g. 2555 for SOX-style controls)."
  default     = 7
}

# arm64 build of the AWS Parameters and Secrets Lambda Extension, pinned to eu-west-1.
variable "secrets_extension_layer_arn" {
  type        = string
  description = "ARN of the AWS Parameters and Secrets Lambda Extension layer."
  default     = "arn:aws:lambda:eu-west-1:015030872274:layer:AWS-Parameters-and-Secrets-Lambda-Extension-Arm64:87"
}

variable "logfire_token" {
  type        = string
  description = "Logfire write token. Leave empty to skip the Logfire integration. Set via TF_VAR_logfire_token in .envrc.local."
  default     = ""
  sensitive   = true
}

The Lambda

Two things in lambda.tf worth pointing at. First, the IAM policy mirrors post 1's Bedrock invoke perms, then layers SSM + KMS Decrypt for the Logfire token (gated by dynamic blocks so they only attach when TF_VAR_logfire_token is set) and firehose:PutRecord on the audit stream:

infra/lambda.tf

# trivy:ignore:avd-aws-0057
# Bedrock foundation-model ARNs do not pin to the caller region (the inference
# profile fans out cross-region), and Marketplace subscription actions are
# global by design.
data "aws_iam_policy_document" "lambda_permissions" {
  # Bedrock invocation. Same shape as iam.tf's bedrock_invoke; copied
  # here so the Lambda role is self-contained and does not require
  # the user-role policy to also be attached to the Lambda role.
  statement {
    actions = [
      "bedrock:Converse",
      "bedrock:ConverseStream",
      "bedrock:InvokeModel",
      "bedrock:InvokeModelWithResponseStream",
    ]
    # Bedrock foundation-model ARNs do not pin to the caller region; the
    # inference profile fans out cross-region, so the * region segment is required.
    #trivy:ignore:avd-aws-0057
    resources = [
      aws_bedrock_inference_profile.agent.arn,
      local.system_inference_profile_arn,
      "arn:aws:bedrock:*::foundation-model/${local.bedrock_model_id}",
    ]
  }

  statement {
    actions = [
      "aws-marketplace:Subscribe",
      "aws-marketplace:Unsubscribe",
      "aws-marketplace:ViewSubscriptions",
    ]
    # Marketplace subscription actions are global by design.
    #trivy:ignore:avd-aws-0057
    resources = ["*"]
  }

  # SSM SecureString read for the Logfire token, only when wired.
  dynamic "statement" {
    for_each = local.logfire_token_wired ? [1] : []
    content {
      actions   = ["ssm:GetParameter"]
      resources = [aws_ssm_parameter.logfire_token[0].arn]
    }
  }

  # KMS Decrypt on the AWS-managed SSM key, required to read a
  # SecureString value through the extension. Only when wired.
  dynamic "statement" {
    for_each = local.logfire_token_wired ? [1] : []
    content {
      actions = ["kms:Decrypt"]
      resources = [
        "arn:aws:kms:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:alias/aws/ssm",
      ]
    }
  }

  statement {
    actions = [
      "firehose:PutRecord",
      "firehose:PutRecordBatch",
    ]
    resources = [aws_kinesis_firehose_delivery_stream.audit.arn]
  }
}

Second, the function attaches the AWS Parameters and Secrets Lambda Extension as a layer. The extension caches SSM reads inside the Lambda execution environment, so we pay the SSM read cost at most once per cold start. The archive_file placeholder is a one-line stub that exists only to let Terraform create the function on first apply; the deploy flow below replaces it. lifecycle.ignore_changes keeps later terraform apply runs from clobbering deployed code.

infra/lambda.tf

resource "aws_lambda_function" "agent" {
  function_name = "terraform-pr-agent"
  role          = aws_iam_role.lambda.arn
  runtime       = "python3.13"
  architectures = ["arm64"]
  handler       = "agent.handler.handler"
  timeout       = 60
  memory_size   = 512

  filename         = data.archive_file.placeholder.output_path
  source_code_hash = data.archive_file.placeholder.output_base64sha256

  layers = [var.secrets_extension_layer_arn]

  tracing_config {
    mode = "Active"
  }

  environment {
    variables = merge(
      {
        BEDROCK_INFERENCE_PROFILE_ARN           = aws_bedrock_inference_profile.agent.arn
        BEDROCK_MODEL_ID                        = local.bedrock_model_id
        PARAMETERS_SECRETS_EXTENSION_CACHE_SIZE = "100"
        PARAMETERS_SECRETS_EXTENSION_HTTP_PORT  = "2773"
        FIREHOSE_DELIVERY_STREAM                = aws_kinesis_firehose_delivery_stream.audit.name
      },
      local.logfire_token_wired ? {
        LOGFIRE_TOKEN_PARAMETER = local.logfire_token_parameter_name
      } : {},
    )
  }

  lifecycle {
    ignore_changes = [
      filename,
      source_code_hash,
    ]
  }

  depends_on = [
    aws_iam_role_policy_attachment.lambda_basic_execution,
    aws_iam_role_policy.lambda_permissions,
  ]
}

KMS and the audit bucket

Our audit bucket gets a dedicated KMS key for SSE-KMS. This allows us to further restrict reads by only allowing certain IAM roles to decrypt the audit bucket's contents. It also gives us another audit trail via CloudTrail: every Decrypt is logged with the IAM role that requested it. We could also add S3 access logging as another layer, but for this post we skip it to keep the moving parts down.

infra/kms.tf

resource "aws_kms_key" "audit" {
  description             = "Encrypts the terraform-pr-agent audit bucket."
  enable_key_rotation     = true
  deletion_window_in_days = 7
  policy                  = data.aws_iam_policy_document.audit_kms_key_resource_policy.json
}

resource "aws_kms_alias" "audit" {
  name          = "alias/terraform-pr-agent-audit"
  target_key_id = aws_kms_key.audit.key_id
}

One common gotcha with KMS is that a key policy (resource policy) is required to enable IAM-based control within the AWS account of the key (or others if you choose). The key policy below does that. Notice the principal: arn:aws:iam::<account>:root is AWS's shorthand for "this account, governed by IAM," not the literal root user. It's how you delegate the key's authorization to IAM policies, so concrete grants can live on role policies instead of in the key policy itself.

infra/kms.tf

data "aws_iam_policy_document" "audit_kms_key_resource_policy" {
  statement {
    sid       = "EnableIAMUserPermissions"
    actions   = ["kms:*"]
    resources = ["*"]
    principals {
      type        = "AWS"
      identifiers = ["arn:aws:iam::${data.aws_caller_identity.current.account_id}:root"]
    }
  }
}

The audit bucket follows a standard setup with versioning enabled and all public access blocked.

infra/audit-bucket.tf

# Access logging would require a second bucket and is out of scope: the audit
# copy here is the system of record for what the agent did, not for who read it.
#trivy:ignore:avd-aws-0089
resource "aws_s3_bucket" "audit" {
  bucket              = local.audit_bucket_name
  object_lock_enabled = true
}

resource "aws_s3_bucket_versioning" "audit" {
  bucket = aws_s3_bucket.audit.id

  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_public_access_block" "audit" {
  bucket                  = aws_s3_bucket.audit.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "aws_s3_bucket_server_side_encryption_configuration" "audit" {
  bucket = aws_s3_bucket.audit.id

  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = aws_kms_key.audit.arn
    }

    bucket_key_enabled = true
  }
}

S3 Object Lock is the actual audit primitive (sometimes called WORM, write-once-read-many). With this config S3 blocks every modification to a locked object, regardless of IAM. We start in GOVERNANCE mode while validating the pipeline; flip to COMPLIANCE once you're confident. The difference matters: in GOVERNANCE you can still delete or shorten retention with s3:BypassGovernanceRetention plus the x-amz-bypass-governance-retention: true header, so mistakes are recoverable. In COMPLIANCE that escape hatch is gone, not even root can shorten the lock. Lifecycle expirations also defer until each object's retention clock runs out (transitions to colder storage classes still work).

For high-risk systems under the EU AI Act, Article 12 requires automatic event logging over the system's lifetime and Article 19 sets a six-month minimum retention. Object Lock enforces that retention at the storage layer rather than via IAM policy, which closes the most common gap in cloud audit trails.

Two things this setup does not give you. First, cryptographic chain-of-custody is not mandated by the Act, but if you ever want tamper-evidence stronger than "the bucket says no", add signed or hash-chained logs on top. Second, GDPR right-to-erasure collides with COMPLIANCE-mode retention: anything personal you write to the bucket is locked in for the full window, so plan what you log about identifiable individuals before turning it on.

infra/audit-bucket.tf

# GOVERNANCE leaves the bucket deletable for the tutorial; switch to "COMPLIANCE" in production.
resource "aws_s3_bucket_object_lock_configuration" "audit" {
  bucket = aws_s3_bucket.audit.id

  rule {
    default_retention {
      mode = "GOVERNANCE"
      days = var.audit_retention_days
    }
  }
}

Simple logic for the lifecycle policy: after the audit retention period ends the current version expires; S3 adds a delete marker, turning the version that held the data into a noncurrent version. One day later that noncurrent version is permanently deleted, so storage stops being billed.

There's also a transition to Glacier IR at day 90, which only fires when audit_retention_days > 90.

infra/audit-bucket.tf

resource "aws_s3_bucket_lifecycle_configuration" "audit" {
  bucket = aws_s3_bucket.audit.id

  rule {
    id     = "transition-to-glacier-ir"
    status = "Enabled"

    filter {}

    transition {
      days          = 90
      storage_class = "GLACIER_IR"
    }

    noncurrent_version_transition {
      noncurrent_days = 90
      storage_class   = "GLACIER_IR"
    }
  }

  rule {
    id     = "expire-after-retention"
    status = "Enabled"

    filter {}

    expiration {
      days = var.audit_retention_days
    }

    noncurrent_version_expiration {
      noncurrent_days = 1
    }
  }
}

Firehose

The agent runs in Lambda, so in principle it can scale to high concurrency. Firehose handles that cleanly: 100k records/second is far more than we should ever need, and it also handles batching: multiple traces (PutRecord calls) get combined into a single S3 object, avoiding the small-files problem on read.

infra/firehose.tf

resource "aws_kinesis_firehose_delivery_stream" "audit" {
  name        = "terraform-pr-agent-audit"
  destination = "extended_s3"

  extended_s3_configuration {
    role_arn            = aws_iam_role.firehose.arn
    bucket_arn          = aws_s3_bucket.audit.arn
    prefix              = "traces/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/"
    error_output_prefix = "errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
    buffering_size      = 5
    buffering_interval  = 60
    compression_format  = "GZIP"

    cloudwatch_logging_options {
      enabled         = true
      log_group_name  = aws_cloudwatch_log_group.firehose.name
      log_stream_name = aws_cloudwatch_log_stream.firehose_s3.name
    }
  }
}

Firehose needs IAM to write encrypted objects into the audit bucket. The only slightly surprising permission is s3:PutObjectRetention. S3 evaluates the lock policy on every PutObject and rejects writers that can't set retention, even when the value being applied is just the bucket default. That detail surfaces another important fact: the lock period is persisted at write time, so changing the bucket default later only affects future objects, not existing ones. This is by design - anything else would undermine the lock.

infra/firehose.tf

data "aws_iam_policy_document" "firehose_assume" {
  statement {
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["firehose.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "firehose" {
  name               = "terraform-pr-agent-firehose"
  assume_role_policy = data.aws_iam_policy_document.firehose_assume.json
}

data "aws_iam_policy_document" "firehose_permissions" {
  statement {
    actions = [
      "s3:PutObject",
      "s3:PutObjectRetention",
      "s3:GetBucketLocation",
      "s3:ListBucket",
    ]
    # /* is how S3 IAM grants object-scoped actions; bounded to the audit bucket.
    #trivy:ignore:avd-aws-0057
    resources = [
      aws_s3_bucket.audit.arn,
      "${aws_s3_bucket.audit.arn}/*",
    ]
  }

  statement {
    actions = [
      "kms:GenerateDataKey",
      "kms:Decrypt",
    ]
    resources = [aws_kms_key.audit.arn]
  }

  statement {
    actions = ["logs:PutLogEvents"]
    # :* covers log streams inside the named Firehose log group only.
    #trivy:ignore:avd-aws-0057
    resources = ["${aws_cloudwatch_log_group.firehose.arn}:*"]
  }
}

resource "aws_iam_role_policy" "firehose_permissions" {
  name   = "terraform-pr-agent-firehose-permissions"
  role   = aws_iam_role.firehose.id
  policy = data.aws_iam_policy_document.firehose_permissions.json
}

Build and deploy

The toy chat.py was a PEP 723 inline script; for Lambda we switch to a proper pyproject.toml:

pyproject.toml

[project]
name = "terraform-pr-agent"
version = "0.1.0"
description = "The terraform-pr-agent Lambda handler."
requires-python = ">=3.13"
dependencies = [
    "pydantic-ai-slim[bedrock]>=1.106,<2",
    "logfire>=4.35,<5",
    "boto3>=1.35,<2",
]

The build script is the Astral uv-on-Lambda flow: export the locked deps to a requirements.txt, install them into a target dir with the manylinux2014 platform pinned, drop agent/ alongside, zip.

scripts/build-lambda.sh

#!/usr/bin/env bash
# Build a deployable zip for the terraform-pr-agent Lambda.
#
# Runs from anywhere; cd's to the project root (the dir holding
# pyproject.toml). Produces build/lambda.zip ready for:
#
#   aws lambda update-function-code \
#       --function-name terraform-pr-agent \
#       --zip-file fileb://build/lambda.zip
#
# Mirrors the pattern from
# https://docs.astral.sh/uv/guides/integration/aws-lambda/
set -euo pipefail

cd "$(dirname "$0")/.."

# Make sure the lock is in sync with pyproject.toml before exporting.
uv sync --quiet

rm -rf build
mkdir -p build/packages

# Export the locked dependency set so uv pip install can consume it
# without re-resolving.
uv export --frozen --no-dev --no-editable -o build/requirements.txt

# Install deps for Lambda's runtime. --python-platform forces wheels
# compatible with Amazon Linux 2 on arm64 (matches the function's
# architectures = ["arm64"]). --no-compile-bytecode keeps the zip small
# and avoids spending cold-start cycles on pyc creation.
uv pip install \
    --no-installer-metadata \
    --no-compile-bytecode \
    --python-platform aarch64-manylinux2014 \
    --python 3.13 \
    --target build/packages \
    -r build/requirements.txt

# Drop the handler package alongside the installed deps.
cp -r agent build/packages/

# Zip from inside the staging dir so paths sit at the zip root.
( cd build/packages && zip -qr ../lambda.zip . )

echo "built: $(pwd)/build/lambda.zip ($(du -h build/lambda.zip | cut -f1))"

Push the zip and smoke-test:

Build, deploy, invoke

chmod +x scripts/build-lambda.sh
./scripts/build-lambda.sh
aws lambda update-function-code \
  --function-name terraform-pr-agent \
  --zip-file fileb://build/lambda.zip
aws lambda invoke \
  --function-name terraform-pr-agent \
  --payload '{"prompt": "Hello"}' \
  /tmp/out.json
cat /tmp/out.json

If TF_VAR_logfire_token was set, the trace lands in your Logfire project. If not, the call still succeeds; logfire just does not ship anywhere.

Wiring pydantic-ai to OTLP

We use the Logfire and OTLP SDKs to wire up the pydantic-ai instrumentation and export into two sinks:

S3 (for audit)
Logfire (for traces, optional)

Fetching the Logfire token at first invoke

Lambda secrets are often wired directly into environment variables. AWS recommends against this: env var values are visible to anyone with read access on the Lambda's config. Instead, Lambdas should use the Parameters and Secrets Lambda Extension to read the token from SSM Parameter Store on the first invocation. The utility function below does that fetch; the layer itself is declared in infra/lambda.tf.

agent/handler.py

def _fetch_logfire_token() -> str | None:
    """Read the Logfire token from the Parameters and Secrets extension.

    Returns None when no token parameter is configured, so the function
    runs fine without the Logfire integration. The extension caches the
    value across invocations, so this is cheap on warm starts.
    """
    parameter_name = os.environ.get("LOGFIRE_TOKEN_PARAMETER")
    if not parameter_name:
        return None

    session_token = os.environ["AWS_SESSION_TOKEN"]

    # safe="" forces urllib to percent-encode the leading and embedded
    # slashes in a hierarchical parameter name (e.g. /a/b/c -> %2Fa%2Fb%2Fc).
    # The Parameters and Secrets extension rejects unencoded slashes with
    # HTTP 400; AWS' own Python sample shows the same %2F-encoded form.
    url = (
        "http://localhost:2773/systemsmanager/parameters/get"
        f"?name={urllib.parse.quote(parameter_name, safe='')}&withDecryption=true"
    )
    req = urllib.request.Request(
        url,
        headers={"X-Aws-Parameters-Secrets-Token": session_token},
    )
    with urllib.request.urlopen(req, timeout=2) as resp:
        payload = json.load(resp)
    return payload["Parameter"]["Value"]

During the first invocation we fetch the token and instrument Logfire. This can't happen at module load time, because the Parameters and Secrets Lambda Extension is not available during Lambda's INIT phase (when the module is imported).

agent/handler.py

@cache
def _init_logfire() -> None:
    """Wire Logfire and the audit span processor once per warm
    container, on the first INVOKE.

    The Parameters and Secrets extension is not ready to serve traffic
    during the Lambda INIT phase, so the token fetch (and the matching
    logfire setup) cannot run at module import time. @cache memoises
    on the empty argument tuple, so this runs exactly once per
    container and is a no-op on every subsequent invocation.
    """
    token = _fetch_logfire_token()
    if token:
        os.environ["LOGFIRE_TOKEN"] = token
    # head=1.0 and tail=None are today's Logfire defaults; pinned here
    # because this is an audit pipeline, so every trace must reach S3.
    # Volume is low (one trace per Lambda invocation) and the audit
    # requirement outweighs Logfire ingest cost. Splitting the rates
    # (e.g. 1% to Logfire, 100% to S3) is possible with a small extra
    # sampler; see the post.
    logfire.configure(
        send_to_logfire="if-token-present",
        sampling=SamplingOptions(head=1.0, tail=None),
        additional_span_processors=[PerTraceAuditProcessor(_ship_trace)],
    )
    # include_content=True is the pydantic-ai default; pinned because the
    # audit copy needs the actual prompts, tool args, and responses to be
    # useful for after-the-fact forensics. If that ever becomes a
    # compliance problem (PII, secrets in prompts), flip to False and
    # accept a metadata-only audit trail.
    logfire.instrument_pydantic_ai(version=5, include_content=True)

The per-trace audit processor

OTel has no OnTraceComplete hook; the only signals are on_start and on_end per span. The processor implements the missing primitive against span.parent is None (root) and a trace_id-keyed buffer - Logfire's tail sampler buffers spans by trace_id the same way internally. We do not use SimpleSpanProcessor + InMemorySpanExporter because the buffer would be module-scoped and might leak across invocations on a warm Lambda container.

ℹ️ Note

Sampling is pinned at 100% in _init_logfire because the audit copy needs every trace.

agent/handler.py

class PerTraceAuditProcessor(SpanProcessor):
    """Buffer spans by trace_id, ship as one batch when the root ends.

    The OTel SDK has no `OnTraceComplete` hook, so this implements it
    against the only signal available: `on_end` fires synchronously and
    `span.parent is None` on a root. Late children (spans ended on a
    transport thread after the root has already shipped) are dropped,
    mirroring logfire's tail sampler. See pydantic/logfire#1034.
    """

    def __init__(
        self,
        on_trace_complete: Callable[[Sequence[ReadableSpan]], None],
    ) -> None:
        self._on_trace_complete = on_trace_complete
        self._buffers: dict[int, list[ReadableSpan]] = {}
        self._shipped: set[int] = set()
        self._lock = threading.Lock()

    def on_end(self, span: ReadableSpan) -> None:
        if not (span.context and span.context.trace_flags.sampled):
            return
        trace_id = span.context.trace_id
        with self._lock:
            if trace_id in self._shipped:
                return
            self._buffers.setdefault(trace_id, []).append(span)
            if span.parent is not None:
                return
            spans = self._buffers.pop(trace_id)
            self._shipped.add(trace_id)
        self._ship(spans)

    def force_flush(self, timeout_millis: int = 30000) -> bool:
        with self._lock:
            pending = list(self._buffers.values())
            self._shipped.update(self._buffers)
            self._buffers.clear()
        for spans in pending:
            self._ship(spans)
        return True

    def shutdown(self) -> None:
        self.force_flush()

    def _ship(self, spans: Sequence[ReadableSpan]) -> None:
        # Suppress instrumentation around the callback so an instrumented
        # boto3/requests client inside it does not emit a span that
        # re-enters on_end for a sibling trace.
        token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
        try:
            self._on_trace_complete(spans)
        finally:
            detach(token)

Shipping one trace per Firehose record

Shipping the trace is one put_record call on Firehose.

agent/handler.py

_firehose = boto3.client("firehose")
_DELIVERY_STREAM = os.environ["FIREHOSE_DELIVERY_STREAM"]


def _ship_trace(spans: Sequence[ReadableSpan]) -> None:
    """Serialise one trace as OTLP-JSON and ship it as a single Firehose record."""
    payload = json_format.MessageToJson(encode_spans(spans), indent=None) + "\n"
    _firehose.put_record(
        DeliveryStreamName=_DELIVERY_STREAM,
        Record={"Data": payload.encode("utf-8")},
    )

Constructing the Bedrock model

The model config is simpler than in post 1: we use the Lambda execution role directly, no assume_role like chat.py.

agent/handler.py

_INFERENCE_PROFILE_ARN = os.environ["BEDROCK_INFERENCE_PROFILE_ARN"]
_MODEL_ID = os.environ["BEDROCK_MODEL_ID"]


def _build_model() -> BedrockConverseModel:
    return BedrockConverseModel(
        _MODEL_ID,
        settings={"bedrock_inference_profile": _INFERENCE_PROFILE_ARN},
    )

The handler entry

The handler does three things: ensure Logfire is initialised, pull the prompt off the event, and call agent.run_sync. Everything else (audit shipping, X-Ray tracing, error surfacing) runs through the per-trace processor and the Logfire SDK - there is no try/except in the handler; failures propagate as Lambda 5xx, and the audit copy is shipped from the processor's on_end at root close, before the exception unwinds.

agent/handler.py

class HandlerEvent(TypedDict):
    prompt: NotRequired[str]


class HandlerResponse(TypedDict):
    status: str
    output: str


def handler(event: HandlerEvent, context: object) -> HandlerResponse:
    """Lambda entry point.

    Falls back to a default prompt so the function can be smoke-tested
    with an empty payload.

    The audit copy ships from inside PerTraceAuditProcessor.on_end when
    the agent's root span closes, so the handler does not need a finally
    block: a Firehose failure raises on the same thread as agent.run_sync
    and propagates as a Lambda 5xx. A failed agent run also closes its
    root span (with status=ERROR) before the exception unwinds, so the
    partial trace still ships.
    """
    _init_logfire()
    prompt = event.get("prompt", "Say hello.")
    result = agent.run_sync(prompt)
    return {
        "status": "ok",
        "output": str(result.output),
    }

The full file:

agent/handler.py

"""AWS Lambda handler for the terraform-pr-agent.

First invocation:
- Reads the Logfire token from SSM via the Parameters and Secrets Lambda
  Extension at http://localhost:2773 when LOGFIRE_TOKEN_PARAMETER is set
  in the env; sets LOGFIRE_TOKEN so the logfire SDK picks it up.
- Configures logfire with send_to_logfire="if-token-present" and registers
  PerTraceAuditProcessor, a custom OTel SpanProcessor that buffers spans
  by trace_id and ships one OTLP-JSON Firehose record when the trace's
  root span ends.
- Instruments pydantic-ai with version=5 for the spec-compliant span
  names (invoke_agent, chat, execute_tool) and current GenAI semantic
  conventions. Pinned explicitly so the schema readers see in the
  audit copy stays stable across pydantic-ai releases.

The extension's HTTP server rejects requests during INIT with a
"not ready to serve traffic" 400, so this work runs on the first
INVOKE and is memoised with @cache for subsequent warm invocations.

Handler:
- Reads `prompt` from the invocation event.
- Runs the agent synchronously and returns the structured output.

The audit copy lands in Firehose from inside the processor's on_end
when the agent root span closes, so the handler has no flush logic.
A Firehose-side failure raises on the same thread as agent.run_sync
and propagates as a Lambda 5xx; the system-of-record copy is never
silently dropped.
"""

import json
import os
import threading
import urllib.parse
import urllib.request
from collections.abc import Callable, Sequence
from functools import cache
from typing import NotRequired, TypedDict

import boto3
import logfire
from google.protobuf import json_format
from logfire.sampling import SamplingOptions
from opentelemetry.context import (
    _SUPPRESS_INSTRUMENTATION_KEY,
    attach,
    detach,
    set_value,
)
from opentelemetry.exporter.otlp.proto.common._internal.trace_encoder import (
    encode_spans,
)
from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor
from pydantic_ai import Agent
from pydantic_ai.models.bedrock import BedrockConverseModel


def _fetch_logfire_token() -> str | None:
    """Read the Logfire token from the Parameters and Secrets extension.

    Returns None when no token parameter is configured, so the function
    runs fine without the Logfire integration. The extension caches the
    value across invocations, so this is cheap on warm starts.
    """
    parameter_name = os.environ.get("LOGFIRE_TOKEN_PARAMETER")
    if not parameter_name:
        return None

    session_token = os.environ["AWS_SESSION_TOKEN"]

    # safe="" forces urllib to percent-encode the leading and embedded
    # slashes in a hierarchical parameter name (e.g. /a/b/c -> %2Fa%2Fb%2Fc).
    # The Parameters and Secrets extension rejects unencoded slashes with
    # HTTP 400; AWS' own Python sample shows the same %2F-encoded form.
    url = (
        "http://localhost:2773/systemsmanager/parameters/get"
        f"?name={urllib.parse.quote(parameter_name, safe='')}&withDecryption=true"
    )
    req = urllib.request.Request(
        url,
        headers={"X-Aws-Parameters-Secrets-Token": session_token},
    )
    with urllib.request.urlopen(req, timeout=2) as resp:
        payload = json.load(resp)
    return payload["Parameter"]["Value"]




class PerTraceAuditProcessor(SpanProcessor):
    """Buffer spans by trace_id, ship as one batch when the root ends.

    The OTel SDK has no `OnTraceComplete` hook, so this implements it
    against the only signal available: `on_end` fires synchronously and
    `span.parent is None` on a root. Late children (spans ended on a
    transport thread after the root has already shipped) are dropped,
    mirroring logfire's tail sampler. See pydantic/logfire#1034.
    """

    def __init__(
        self,
        on_trace_complete: Callable[[Sequence[ReadableSpan]], None],
    ) -> None:
        self._on_trace_complete = on_trace_complete
        self._buffers: dict[int, list[ReadableSpan]] = {}
        self._shipped: set[int] = set()
        self._lock = threading.Lock()

    def on_end(self, span: ReadableSpan) -> None:
        if not (span.context and span.context.trace_flags.sampled):
            return
        trace_id = span.context.trace_id
        with self._lock:
            if trace_id in self._shipped:
                return
            self._buffers.setdefault(trace_id, []).append(span)
            if span.parent is not None:
                return
            spans = self._buffers.pop(trace_id)
            self._shipped.add(trace_id)
        self._ship(spans)

    def force_flush(self, timeout_millis: int = 30000) -> bool:
        with self._lock:
            pending = list(self._buffers.values())
            self._shipped.update(self._buffers)
            self._buffers.clear()
        for spans in pending:
            self._ship(spans)
        return True

    def shutdown(self) -> None:
        self.force_flush()

    def _ship(self, spans: Sequence[ReadableSpan]) -> None:
        # Suppress instrumentation around the callback so an instrumented
        # boto3/requests client inside it does not emit a span that
        # re-enters on_end for a sibling trace.
        token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
        try:
            self._on_trace_complete(spans)
        finally:
            detach(token)




_firehose = boto3.client("firehose")
_DELIVERY_STREAM = os.environ["FIREHOSE_DELIVERY_STREAM"]


def _ship_trace(spans: Sequence[ReadableSpan]) -> None:
    """Serialise one trace as OTLP-JSON and ship it as a single Firehose record."""
    payload = json_format.MessageToJson(encode_spans(spans), indent=None) + "\n"
    _firehose.put_record(
        DeliveryStreamName=_DELIVERY_STREAM,
        Record={"Data": payload.encode("utf-8")},
    )




@cache
def _init_logfire() -> None:
    """Wire Logfire and the audit span processor once per warm
    container, on the first INVOKE.

    The Parameters and Secrets extension is not ready to serve traffic
    during the Lambda INIT phase, so the token fetch (and the matching
    logfire setup) cannot run at module import time. @cache memoises
    on the empty argument tuple, so this runs exactly once per
    container and is a no-op on every subsequent invocation.
    """
    token = _fetch_logfire_token()
    if token:
        os.environ["LOGFIRE_TOKEN"] = token
    # head=1.0 and tail=None are today's Logfire defaults; pinned here
    # because this is an audit pipeline, so every trace must reach S3.
    # Volume is low (one trace per Lambda invocation) and the audit
    # requirement outweighs Logfire ingest cost. Splitting the rates
    # (e.g. 1% to Logfire, 100% to S3) is possible with a small extra
    # sampler; see the post.
    logfire.configure(
        send_to_logfire="if-token-present",
        sampling=SamplingOptions(head=1.0, tail=None),
        additional_span_processors=[PerTraceAuditProcessor(_ship_trace)],
    )
    # include_content=True is the pydantic-ai default; pinned because the
    # audit copy needs the actual prompts, tool args, and responses to be
    # useful for after-the-fact forensics. If that ever becomes a
    # compliance problem (PII, secrets in prompts), flip to False and
    # accept a metadata-only audit trail.
    logfire.instrument_pydantic_ai(version=5, include_content=True)




_INFERENCE_PROFILE_ARN = os.environ["BEDROCK_INFERENCE_PROFILE_ARN"]
_MODEL_ID = os.environ["BEDROCK_MODEL_ID"]


def _build_model() -> BedrockConverseModel:
    return BedrockConverseModel(
        _MODEL_ID,
        settings={"bedrock_inference_profile": _INFERENCE_PROFILE_ARN},
    )




SYSTEM_PROMPT = (
    "You are the terraform-pr-agent. For now you are a placeholder; "
    "respond briefly to whatever prompt you are given."
)


agent = Agent(_build_model(), system_prompt=SYSTEM_PROMPT)


class HandlerEvent(TypedDict):
    prompt: NotRequired[str]


class HandlerResponse(TypedDict):
    status: str
    output: str


def handler(event: HandlerEvent, context: object) -> HandlerResponse:
    """Lambda entry point.

    Falls back to a default prompt so the function can be smoke-tested
    with an empty payload.

    The audit copy ships from inside PerTraceAuditProcessor.on_end when
    the agent's root span closes, so the handler does not need a finally
    block: a Firehose failure raises on the same thread as agent.run_sync
    and propagates as a Lambda 5xx. A failed agent run also closes its
    root span (with status=ERROR) before the exception unwinds, so the
    partial trace still ships.
    """
    _init_logfire()
    prompt = event.get("prompt", "Say hello.")
    result = agent.run_sync(prompt)
    return {
        "status": "ok",
        "output": str(result.output),
    }

Querying traces

DuckDB on your laptop

The audit copy is GZIP'd OTLP-JSON under a Hive-partitioned prefix. Anything that reads NDJSON over S3 can query it; for a single-laptop workflow the lightest option is DuckDB. Install it once (brew install duckdb on macOS; see the install guide for other platforms).

The view reads the bucket name from AUDIT_BUCKET in your shell (via DuckDB's CLI-only getenv()), so park that alongside AWS_PROFILE in .envrc.local and let direnv export both on every cd:

.envrc.local

# Bucket name for the audit copy. The DuckDB view in scripts/traces.sql
# reads it via the CLI-only getenv() function so the SQL stays free of
# account-specific values.
export AUDIT_BUCKET=terraform-pr-agent-audit-$(aws sts get-caller-identity --query Account --output text)-${AWS_REGION}

Then opening the local database file is a one-liner:

duckdb traces.duckdb

First session only, create a persistent S3 secret and a view that flattens the OTLP shape into the columns you actually want. Paste this in (or .read it from a saved file):

scripts/traces.sql

CREATE PERSISTENT SECRET (
    TYPE s3,
    PROVIDER credential_chain,
    REFRESH auto
);
SET VARIABLE audit_bucket = getenv('AUDIT_BUCKET');

-- hive_partitioning = true reads year=YYYY/month=MM/day=DD/ from the
-- object path as virtual columns, so the partition predicate in a
-- query below prunes objects before any file is opened.
CREATE OR REPLACE VIEW traces AS
WITH spans AS (
    -- Flatten the OTLP-JSON envelope into one row per span, with the
    -- common span fields lifted out as named columns so downstream
    -- CTEs and ad-hoc queries can work against `name`, `trace_id`,
    -- `dur_ms`, etc. without re-doing the struct navigation each time.
    SELECT
        year, month, day,
        span.name                                                                AS name,
        lower(hex(from_base64(span.traceId::VARCHAR)))                           AS trace_id,
        lower(hex(from_base64(span.spanId::VARCHAR)))                            AS span_id,
        lower(hex(from_base64(span.parentSpanId::VARCHAR)))                      AS parent_span_id,
        make_timestamp_ns(span.startTimeUnixNano::BIGINT)                        AS started,
        make_timestamp_ns(span.endTimeUnixNano::BIGINT)                          AS ended,
        (span.endTimeUnixNano::BIGINT - span.startTimeUnixNano::BIGINT) / 1e6    AS dur_ms,
        span.status.code::VARCHAR                                                AS status_code,
        span.attributes                                                          AS attributes,
        data.filename                                                            AS source_file,
        -- Firehose names objects <stream>-<ver>-<YYYY-MM-DD-HH-MM-SS>-<uuid>.gz;
        -- stripping the trailing -<uuid>.gz collapses rows from the same flush
        -- batch onto a stable key for grouping.
        regexp_replace(
            split_part(data.filename, '/', -1),
            '-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.gz$',
            ''
        )                                                                        AS batch_key
    FROM read_ndjson(
        's3://' || getvariable('audit_bucket') || '/traces/**/*.gz',
        compression = 'gzip', hive_partitioning = true, filename = true) AS data
       , UNNEST(data.resourceSpans) AS u1(rs)
       , UNNEST(rs.scopeSpans)      AS u2(ss)
       , UNNEST(ss.spans)           AS u3(span)
),
roots AS (
    -- One row per trace: the invoke_agent root span pydantic-ai emits per run.
    SELECT * FROM spans
    WHERE parent_span_id IS NULL OR parent_span_id = ''
),
chats AS (
    -- Per-trace summary of the LLM call: pulls the GenAI semantic
    -- convention attributes off the chat span and exposes each as a
    -- named column.
    SELECT
        trace_id,
        list_filter(attributes, x -> x.key = 'gen_ai.system')[1].value.stringValue                     AS gen_ai_system,
        list_filter(attributes, x -> x.key = 'gen_ai.operation.name')[1].value.stringValue             AS operation,
        list_filter(attributes, x -> x.key = 'gen_ai.request.model')[1].value.stringValue              AS request_model,
        list_filter(attributes, x -> x.key = 'gen_ai.response.model')[1].value.stringValue             AS response_model,
        list_filter(attributes, x -> x.key = 'gen_ai.usage.input_tokens')[1].value.intValue::BIGINT    AS in_tokens,
        list_filter(attributes, x -> x.key = 'gen_ai.usage.output_tokens')[1].value.intValue::BIGINT   AS out_tokens,
        list_filter(attributes, x -> x.key = 'gen_ai.response.finish_reasons')[1]
            .value.arrayValue.values[1].stringValue                                                    AS finish,
        list_filter(attributes, x -> x.key = 'gen_ai.conversation.id')[1].value.stringValue            AS conversation_id,
        list_filter(attributes, x -> x.key = 'gen_ai.agent.name')[1].value.stringValue                 AS agent_name,
        list_filter(attributes, x -> x.key = 'gen_ai.agent.call.id')[1].value.stringValue              AS agent_call_id,
        list_filter(attributes, x -> x.key = 'gen_ai.input.messages')[1].value.stringValue             AS input_messages,
        list_filter(attributes, x -> x.key = 'gen_ai.output.messages')[1].value.stringValue            AS output_messages
    FROM spans
    WHERE name LIKE 'chat %'
)
SELECT
    roots.started,
    roots.year, roots.month, roots.day,
    roots.trace_id,
    substr(roots.trace_id, 1, 8)                       AS trace,
    roots.batch_key,
    roots.source_file,
    roots.dur_ms,
    regexp_extract(chats.request_model, '[^./]+$')     AS model,
    chats.in_tokens,
    chats.out_tokens,
    chats.finish,
    chats.agent_name,
    chats.conversation_id,
    -- Convenience columns for the common single-turn shape: system at
    -- input[0], user at input[1], assistant at output[0]. Multi-turn
    -- runs invalidate the indices, so reach for input_messages and
    -- output_messages directly for those.
    json_extract_string(chats.input_messages,  '$[0].parts[0].content')  AS system_prompt,
    json_extract_string(chats.input_messages,  '$[1].parts[0].content')  AS user_prompt,
    json_extract_string(chats.output_messages, '$[0].parts[0].content')  AS assistant_response,
    chats.input_messages,
    chats.output_messages,
    -- Per the OTel spec, instrumentation libraries leave status unset
    -- on success (only application code may set it to Ok). Every OTel
    -- backend treats unset as "no error reported"; we render the same.
    CASE roots.status_code
        WHEN 'STATUS_CODE_ERROR' THEN 'err'
        ELSE 'ok'
    END                                                AS status
FROM roots
JOIN chats USING (trace_id);

After that, every later duckdb traces.duckdb session lands you in a shell where the view is already there. One caveat: the view body references getvariable('audit_bucket'), and SET VARIABLE is session-scoped, so a fresh session needs the variable re-set before any SELECT against traces will run. The simplest fix is to .read scripts/traces.sql again at the top of each session - CREATE OR REPLACE VIEW is idempotent, and the same script sets the variable. Hive partitioning means year, month, and day are real columns DuckDB prunes on before opening any object:

Today's runs

SELECT *
FROM traces
WHERE year = 2026
  AND month = 6
  AND day = 8
ORDER BY started DESC;

-- Example output (narrowed to 7 of the view's 18 columns; SELECT * also
-- returns conversation_id, system_prompt, user_prompt, assistant_response,
-- input_messages, output_messages, status, batch_key, source_file,
-- year/month/day):
--
-- ┌───────────────────────────────┬──────────┬────────────────────────────────┬─────────────┬───────────┬────────────┬────────┐
-- │            started            │  trace   │             model              │   dur_ms    │ in_tokens │ out_tokens │ finish │
-- ├───────────────────────────────┼──────────┼────────────────────────────────┼─────────────┼───────────┼────────────┼────────┤
-- │ 2026-06-08 20:31:06.044709547 │ 019ea8ee │ claude-haiku-4-5-20251001-v1:0 │ 6815.733257 │ 36        │ 59         │ stop   │
-- │ 2026-06-08 20:31:05.984812728 │ 019ea8ee │ claude-haiku-4-5-20251001-v1:0 │ 1459.186377 │ 36        │ 38         │ stop   │
-- │ 2026-06-08 20:31:05.964196348 │ 019ea8ee │ claude-haiku-4-5-20251001-v1:0 │ 2024.133412 │ 36        │ 123        │ stop   │
-- │ 2026-06-08 20:31:05.944446527 │ 019ea8ee │ claude-haiku-4-5-20251001-v1:0 │ 1581.984687 │ 36        │ 50         │ stop   │
-- │ 2026-06-08 20:31:05.926721468 │ 019ea8ee │ claude-haiku-4-5-20251001-v1:0 │ 1480.496671 │ 36        │ 43         │ stop   │
-- └───────────────────────────────┴──────────┴────────────────────────────────┴─────────────┴───────────┴────────────┴────────┘

Runs per day

SELECT day,
       count(*) AS runs
FROM traces
WHERE year = 2026
  AND month = 6
GROUP BY day
ORDER BY day;

-- Example output:
--
-- ┌─────┬──────┐
-- │ day │ runs │
-- ├─────┼──────┤
-- │ 04  │ 2    │
-- │ 08  │ 11   │
-- │ 09  │ 1    │
-- └─────┴──────┘

Tokens per day

SELECT day,
       sum(in_tokens)  AS input_tokens,
       sum(out_tokens) AS output_tokens
FROM traces
WHERE year = 2026
GROUP BY day
ORDER BY day;

-- Example output:
--
-- ┌─────┬──────────────┬───────────────┐
-- │ day │ input_tokens │ output_tokens │
-- ├─────┼──────────────┼───────────────┤
-- │ 04  │ 80           │ 43            │
-- │ 08  │ 397          │ 605           │
-- │ 09  │ 38           │ 145           │
-- └─────┴──────────────┴───────────────┘

The view is the interface; everything downstream is normal SQL against a traces table that happens to live in S3. The partition predicates are the same shape the Athena section below uses.

The local traces.duckdb only holds the view DDL and the cached S3 secret. Trace data is streamed from the bucket on every SELECT via httpfs, not copied locally.

Athena/Snowflake

DuckDB is the simplest choice for the post, but every query pulls bytes to your laptop. Because the bucket is Hive-partitioned, the same query shape works in Athena or Snowflake with no schema rebuild.

Reference:

AWS docs: Athena partition projection for Firehose data. CREATE EXTERNAL TABLE with projection.* properties and a storage.location.template infers partitions from the year=YYYY/month=MM/day=DD/hour=HH/ path at query time, so there is no Glue crawler to schedule or partitions to register as new objects land.
Snowflake docs: creating an S3 stage. Create an external stage defining JSON format.

The same trace in Logfire

The same data as above can be viewed independently in the Logfire SaaS platform if you signed up and created a token. You can see in the screenshot that System Prompt, User Prompt, and Output of the agent have been captured.

End state

Every agent run lands in S3 as an immutable OTel trace, with full input/output messages, tool calls, retries, and token usage captured. Athena queries the audit copy when you need to answer compliance questions; Logfire renders the same trace live when you need to debug. The next four posts build retries, conventions, evals, and the PR flow on top of these traces; everything they do shows up in the trace tree without further audit plumbing.

Coming next: ~~workspace and small toolkit for the agent to get to work.~~ Edit: European foundational model capabilities.

Bedrock and pydantic-ai with observability from day one

Andreas Lang — Tue, 16 Jun 2026 11:30:34 +0000

This is the first post in Terraform PR Agent, a series that builds an AI agent which writes Terraform and opens review-ready pull requests, running on AWS Bedrock with pydantic-ai. Later posts add observability, file tools and a validate loop, conventions and policy, evals for choosing a model with numbers, and the guardrails to ship it against a real repo. The posts share one project scaffold, so the prerequisites below are a one-time setup for the whole series.

Prerequisites

Tooling and AWS access common to every post in this series.

Tooling

Terraform 1.x (install). Every post provisions infrastructure with Terraform.
uv for Python project management (install). Each post ships a runnable script you can invoke with uv run.
direnv (install) so terraform, uv run, and aws pick up AWS credentials automatically on cd. The project scaffold ships an .envrc that sources a gitignored .envrc.local.
(Optional) A coding agent such as Claude Code, Cursor, Codex, or Gemini CLI to consume the AgentPrompt blocks throughout the series. Not required (each prompt has a manual equivalent shown alongside it), but it skips the boilerplate.

🤖 Agent prompt: Check and install missing tooling

You are helping set up tooling for a tutorial project.

For each of `terraform`, `uv`, and `direnv`, run `command -v` to
check whether it is installed. If present, print the version and
continue.

For missing tools, detect the system package manager in this order:
`command -v brew`, `command -v dnf`, `command -v apt-get`. Use the
first one available:

  - Terraform: `brew tap hashicorp/tap && brew install hashicorp/tap/terraform`,
    dnf via the HashiCorp RPM repo, or apt via the HashiCorp deb repo.
  - uv: `brew install uv`, or the official installer
    `curl -LsSf https://astral.sh/uv/install.sh | sh`.
  - direnv: `brew install direnv`, `dnf install direnv`, or
    `apt-get install direnv`.

If no package manager is available or the install fails, stop and
link the manual install page so the developer can finish by hand:

  - Terraform: https://developer.hashicorp.com/terraform/install
  - uv: https://docs.astral.sh/uv/getting-started/installation/
  - direnv: https://direnv.net/docs/installation.html

After installing direnv, do not modify any shell rc files. Print the
hook line for the developer's shell (bash, zsh, or fish) and the path
to the relevant rc file, then wait for them to apply it themselves.

Report which tools were already present, which you installed, and
which need manual follow-up.

AWS access

A sandbox, test, or personal AWS account with permission to create, modify, and delete the resources discussed in each post. If you don't have one, follow the official Create Your AWS Account walkthrough (about ten minutes; requires a credit card and a phone number for verification). Treat it as disposable - you can close it from the billing console after the series.
AWS credentials available locally via aws configure sso, aws configure, or whichever method matches your setup. You wire them into the project through .envrc.local in the next section, not your shell rc.

Anthropic First Time Use

Bedrock requires a one-time use-case form per account (or per AWS Organization management account) before Anthropic models can be invoked. Easiest path: open any Claude model in the Bedrock console playground and submit the form. Auto-subscription on first invoke can take up to 15 minutes to settle, so it is worth clearing this before post 1.

CLI alternative and verification

Programmatic equivalent (requires AWS CLI 2.27.42 or later):

aws bedrock put-use-case-for-model-access \
  --form-data "$(printf '{"companyName":"...","companyWebsite":"...","intendedUsers":"1","industryOption":"...","otherIndustryOption":"","useCases":"..."}' | base64)"

Verify:

aws bedrock get-foundation-model-availability \
  --model-id anthropic.claude-haiku-4-5-20251001-v1:0 \
  --region eu-west-1

Look for agreementAvailability.status: AVAILABLE. Expected output:

{
  "modelId": "anthropic.claude-haiku-4-5-20251001-v1",
  "agreementAvailability": { "status": "AVAILABLE" },
  "authorizationStatus": "AUTHORIZED",
  "entitlementAvailability": "AVAILABLE",
  "regionAvailability": "AVAILABLE"
}

If the form has not been submitted, only agreementAvailability.status flips to NOT_AVAILABLE. The other three fields stay green even when invocation would fail, so do not rely on them.

Project scaffold

Download and unpack the scaffold into your projects directory (swap ~/projects for whatever location you prefer):

mkdir -p ~/projects
cd ~/projects
curl -fsSL https://andreaslang.dev/terraform-pr-agent/terraform-pr-agent.tar.gz | tar xz

You should now have:

terraform-pr-agent/
  infra/         Terraform (Bedrock, CloudWatch, DynamoDB, ...)
  agent/         Python package (pydantic-ai code)
  evals/         Golden cases + eval harness
  scripts/       Single-file Python scripts (PEP 723 inline deps, run via `uv run`)
  AGENTS.md      Conventions for AI coding agents (see below)
  .envrc         Sources .envrc.local for direnv
  .envrc.local   Your AWS env (gitignored, fill in locally)
  .gitignore     Excludes .envrc.local, .terraform/, state files

This is the base scaffold; subsequent posts in the series add and modify files inside this tree.

Or have an agent do the same:

🤖 Agent prompt: Fetch and unpack the project scaffold

You are helping set up a tutorial project on a developer's machine.

Fetch and unpack the scaffold tarball into ~/projects/:

  mkdir -p ~/projects
  cd ~/projects
  curl -fsSL \
    https://andreaslang.dev/terraform-pr-agent/terraform-pr-agent.tar.gz \
    | tar xz

Verify with `ls -la ~/projects/terraform-pr-agent` and report any
failures. Do not install dependencies, do not run `direnv allow`,
and do not fill in any values in .envrc.local: the developer will
do that themselves in the next step.

Edit .envrc.local, uncomment Option A (named profile) or Option B (static / temporary creds), then run direnv allow . in the project root. The template:

.envrc.local

# Local AWS env for this project. Gitignored.
# Set the region (shared by both auth options) and uncomment ONE of the
# credential blocks below, then run `direnv allow` in this directory.

export AWS_REGION=eu-west-1

# Option A: Named profile (e.g. from `aws configure sso` or ~/.aws/credentials).
# export AWS_PROFILE=your-profile-name

# Option B: Static or temporary credentials.
# export AWS_ACCESS_KEY_ID=...
# export AWS_SECRET_ACCESS_KEY=...
# export AWS_SESSION_TOKEN=...   # only if using temporary creds

📁 What else is in the scaffold

.envrc

source_env_if_exists .envrc.local

.gitignore

.envrc.local
.terraform/
*.tfstate
*.tfstate.*
.direnv/

The AGENTS.md template is tool-agnostic: Claude Code, Cursor, Codex, and Gemini CLI all read it automatically at session start.

AGENTS.md

# Project conventions

Tutorial project from the Terraform PR Agent series at
https://andreaslang.dev/posts/terraform-pr-agent/

> When a post introduces a new top-level directory, ship an updated
> `AGENTS.md` in that post's `scaffold/` overlay so this Layout section
> stays accurate.

## Layout

- `infra/` Terraform (Bedrock, CloudWatch, DynamoDB, ...)
- `agent/` Python package (pydantic-ai code)
- `evals/` Golden cases + eval harness
- `scripts/` Standalone single-file Python scripts with PEP 723 inline
  dependency metadata. Run with `uv run scripts/<name>.py`; do not move
  these into the `agent` package or a shared `pyproject.toml`.

## Tooling

- Python: use `uv` for dependency and script management. Run scripts with
  `uv run`. Single-file scripts under `scripts/` declare their deps inline
  via PEP 723 (`# /// script ... # ///` headers) so they stay
  self-contained and runnable without a project venv.
- Terraform: 1.x.
- AWS: credentials configured via `aws configure sso` or static keys.

## Conventions

- Run `terraform validate` after editing any `.tf` file.
- Run `terraform fmt` before committing `.tf`.
- Never auto-apply Terraform; print the plan first and wait for confirmation.
- Don't introduce dependencies the current post hasn't covered.
- AWS resources live in a sandbox sub-account; never assume a production account.

What this post covers

We stand up Bedrock + pydantic-ai end to end and put a CloudWatch dashboard and alert on top before writing a single line of agent logic. A local pydantic-ai script validates the stack by invoking Claude Haiku 4.5 through a project-scoped application inference profile, with all the permissions wrapped in a single IAM role.

Most projects only wire monitoring once there is something to monitor. Agentic workloads bite earlier: a small shift in how prompts or data hit the model can turn a previously fine task into many more turns and a much bigger bill, and you do not notice until the invoice lands. Eval (covered in a later post) and monitoring are not optional past hello-world.

See this diagram full-size on the original post.

Bedrock IAM and inference profile

Three AWS resources need to exist before any agent code can run:

The Bedrock inference profile. We could call the model directly via its Bedrock model ID, but an inference profile buys us two things.
- It allows you to switch out models without changing the agent (same inference profile ARN calling a different model)
- It allows you to monitor usage and create alerts (what we need)
An IAM role that we can assume within our AWS account ("arn:aws:iam::${...}:root") to invoke the model and use the Converse API. AWS recommends the Converse API for most chat-style calls, and the pydantic-ai model we use here, BedrockConverseModel, goes through it. You still need the IAM permissions for InvokeModel, though, because that is the action name Converse maps to internally.
Basic terraform main.tf with the provider setup.

The files below drop into the paths shown on each tab; copy them into your own project, then run terraform init and terraform apply. State stays local since there is no backend block - fine for a learning project.

infra/bedrock.tf

locals {
  bedrock_model_id            = "anthropic.claude-haiku-4-5-20251001-v1:0"
  bedrock_cross_region_prefix = "eu"

  system_inference_profile_arn = format(
    "arn:aws:bedrock:%s:%s:inference-profile/%s.%s",
    data.aws_region.current.region,
    data.aws_caller_identity.current.account_id,
    local.bedrock_cross_region_prefix,
    local.bedrock_model_id,
  )
}

resource "aws_bedrock_inference_profile" "agent" {
  name        = "terraform-pr-agent"
  description = "Application inference profile for the Terraform PR Agent series."

  model_source {
    copy_from = local.system_inference_profile_arn
  }
}

output "inference_profile_arn" {
  value = aws_bedrock_inference_profile.agent.arn
}

infra/iam.tf

data "aws_iam_policy_document" "bedrock_invoke" {
  statement {
    actions = [
      "bedrock:Converse",
      "bedrock:ConverseStream",
      "bedrock:InvokeModel",
      "bedrock:InvokeModelWithResponseStream",
    ]
    # Bedrock foundation-model ARNs do not pin to the caller region; the
    # inference profile fans out cross-region, so the * region segment is required.
    #trivy:ignore:avd-aws-0057
    resources = [
      aws_bedrock_inference_profile.agent.arn,
      local.system_inference_profile_arn,
      "arn:aws:bedrock:*::foundation-model/${local.bedrock_model_id}",
    ]
  }

  # Anthropic models on Bedrock are distributed via AWS Marketplace. On the
  # first invocation from a new account, Bedrock auto-subscribes the account
  # to the model product, which requires the invoking principal to hold these
  # Marketplace actions. Once the subscription is active these calls become
  # no-ops, but the principal still needs ViewSubscriptions on every call so
  # Bedrock can confirm the subscription is in place.
  statement {
    actions = [
      "aws-marketplace:Subscribe",
      "aws-marketplace:Unsubscribe",
      "aws-marketplace:ViewSubscriptions",
    ]
    # Marketplace subscription actions are global by design.
    #trivy:ignore:avd-aws-0057
    resources = ["*"]
  }
}

resource "aws_iam_policy" "bedrock_invoke" {
  name        = "terraform-pr-agent-bedrock-invoke"
  description = "Invoke Claude via the terraform-pr-agent inference profile."
  policy      = data.aws_iam_policy_document.bedrock_invoke.json
}

data "aws_iam_policy_document" "agent_assume" {
  statement {
    actions = ["sts:AssumeRole"]
    principals {
      type        = "AWS"
      identifiers = ["arn:aws:iam::${data.aws_caller_identity.current.account_id}:root"]
    }
  }
}

resource "aws_iam_role" "agent" {
  name               = "terraform-pr-agent"
  assume_role_policy = data.aws_iam_policy_document.agent_assume.json
}

resource "aws_iam_role_policy_attachment" "agent_bedrock_invoke" {
  role       = aws_iam_role.agent.name
  policy_arn = aws_iam_policy.bedrock_invoke.arn
}

output "agent_role_arn" {
  value = aws_iam_role.agent.arn
}

infra/main.tf

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "6.17.0"
    }
  }
}

provider "aws" {}

data "aws_caller_identity" "current" {}
data "aws_region" "current" {}

First pydantic-ai call

Now that the infra is in place, let's call the model. Calling Bedrock first means any region, quota, or IAM mistakes surface immediately, before there's any agent code to debug them against. We will do it with a uv run script that uses PEP 723 - Inline script metadata. It allows you to declare the dependencies of a script in metadata and uv will download them for you before running the script. The script we build is a simple console chat interface with the twist that to shut it down we can tell the agent that we want to. This works by enforcing a structured response type, but more on that later.

Add the new env vars to `.envrc.local`

Before we can call anything we need some information about the infra we just set up. In particular the AGENT_ROLE_ARN (we will assume this role in the script) and the BEDROCK_INFERENCE_PROFILE_ARN (we will use this to invoke the model). You can get these by running the terraform output command as shown in the comments of the .envrc.local file.

.envrc.local

# Fill in after `terraform apply` in post 1, then run `direnv reload`.
# Values come from terraform outputs:
#   terraform -chdir=infra output -raw agent_role_arn
#   terraform -chdir=infra output -raw inference_profile_arn
# export AGENT_ROLE_ARN=arn:aws:iam::<account>:role/terraform-pr-agent
# export BEDROCK_INFERENCE_PROFILE_ARN=arn:aws:bedrock:eu-west-1:<account>:application-inference-profile/<id>

Full script

You can copy the full script below into a file called scripts/chat.py. We will look at specific sections below, so keep it at hand.

scripts/chat.py

# /// script
# requires-python = ">=3.11"
# dependencies = [
#   "pydantic-ai-slim[bedrock]>=1.106,<2",
#   "boto3>=1.35,<2",
# ]
# ///
"""Command-line chat against Claude Haiku 4.5 on Bedrock.

Reads three env vars (set them in .envrc.local; see post 1):

  AGENT_ROLE_ARN
    IAM role to assume. Source: terraform -chdir=infra output -raw agent_role_arn
  BEDROCK_INFERENCE_PROFILE_ARN
    Application inference profile that wraps the EU CRIS profile.
    Source: terraform -chdir=infra output -raw inference_profile_arn
  AWS_REGION
    Region for the bedrock-runtime endpoint. Defaults to eu-west-1.

Run with `uv run scripts/chat.py` from the project root.
"""

import os
import sys

import boto3
from botocore.exceptions import ClientError
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.bedrock import BedrockConverseModel
from pydantic_ai.providers.bedrock import BedrockProvider

MODEL_ID = "anthropic.claude-haiku-4-5-20251001-v1:0"

SYSTEM_PROMPT = (
    "You are a conversational assistant running in a small command-line chat. "
    "Respond naturally to the user in the `message` field. "
    "Set `terminate=True` only when the user explicitly asks to end the "
    "conversation (for example: bye, quit, exit, stop, that's all). "
    "When you set `terminate=True`, include a short farewell in `message`. "
    "In every other case set `terminate=False`."
)


class Reply(BaseModel):
    """Structured response on every turn so the loop can decide when to stop."""

    message: str = Field(description="Natural-language reply to the user.")
    terminate: bool = Field(
        description=(
            "True only when the user has explicitly asked to end the conversation; False otherwise."
        )
    )




def require_env(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        sys.stderr.write(
            f"error: {name} is not set. Add it to .envrc.local from "
            f"`terraform output`, then run `direnv reload`.\n"
        )
        sys.exit(2)
    return value


def assume_role(role_arn: str) -> dict[str, str]:
    """Trade the local SSO identity for short-lived terraform-pr-agent creds."""
    sts = boto3.client("sts")
    response = sts.assume_role(RoleArn=role_arn, RoleSessionName="chat-cli")
    return response["Credentials"]




def build_model(role_arn: str, inference_profile_arn: str, region: str) -> BedrockConverseModel:
    creds = assume_role(role_arn)
    bedrock_client = boto3.client(
        "bedrock-runtime",
        region_name=region,
        aws_access_key_id=creds["AccessKeyId"],
        aws_secret_access_key=creds["SecretAccessKey"],
        aws_session_token=creds["SessionToken"],
    )
    # The application inference profile ARN goes in settings, not model_name:
    # pydantic-ai still uses the foundation-model id for capability detection
    # and routes the actual Converse call through the profile.
    return BedrockConverseModel(
        MODEL_ID,
        provider=BedrockProvider(bedrock_client=bedrock_client),
        settings={"bedrock_inference_profile": inference_profile_arn},
    )




def main() -> None:
    role_arn = require_env("AGENT_ROLE_ARN")
    inference_profile_arn = require_env("BEDROCK_INFERENCE_PROFILE_ARN")
    region = os.environ.get("AWS_REGION", "eu-west-1")

    try:
        model = build_model(role_arn, inference_profile_arn, region)
    except ClientError as err:
        sys.stderr.write(
            f"error: could not assume {role_arn}: {err}\n"
            f"check that `terraform apply` has run and AWS_PROFILE is set.\n"
        )
        sys.exit(1)

    agent = Agent(model, output_type=Reply, system_prompt=SYSTEM_PROMPT)

    print('Chatting with Claude Haiku 4.5 via terraform-pr-agent. Type "bye" to exit.')
    history: list = []
    while True:
        try:
            user_input = input("you> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if not user_input:
            continue
        try:
            result = agent.run_sync(user_input, message_history=history)
        except ClientError as err:
            sys.stderr.write(f"bedrock error: {err}\n")
            continue
        reply = result.output
        print(f"agent> {reply.message}")
        history = result.all_messages()
        if reply.terminate:
            break


if __name__ == "__main__":
    main()

System prompt - Basic Chat

Here you can see the system prompt, which sets the scene of the agent's purpose. This is not the system prompt we will use for the agent in our target use case (creating terraform PRs), but just a demo prompt to give you a feel for how the LLM works. Note the specific instructions referencing termination of the program. They reference the Reply schema, which we will define next.

scripts/chat.py

SYSTEM_PROMPT = (
    "You are a conversational assistant running in a small command-line chat. "
    "Respond naturally to the user in the `message` field. "
    "Set `terminate=True` only when the user explicitly asks to end the "
    "conversation (for example: bye, quit, exit, stop, that's all). "
    "When you set `terminate=True`, include a short farewell in `message`. "
    "In every other case set `terminate=False`."
)

Reply schema

Now you see why the tool is called pydantic-ai. The response model (what the LLM has to return) is a pydantic BaseModel. If you know pydantic then you know that for each model you can generate a JSON schema, which can be communicated to the LLM. pydantic-ai picks one of three modes depending on what the model supports. Forced tool call has the LLM call a tool whose schema matches our response model. Native output lets the model return structured data directly, but the model has to support it. Prompted output is the fallback: pydantic-ai injects schema instructions into the prompt and parses the reply. More information here. Tool output with enforced usage is the default.

scripts/chat.py

class Reply(BaseModel):
    """Structured response on every turn so the loop can decide when to stop."""

    message: str = Field(description="Natural-language reply to the user.")
    terminate: bool = Field(
        description=(
            "True only when the user has explicitly asked to end the conversation; False otherwise."
        )
    )

Assume the role

Strictly we do not need to assume a role here - the user profile has plenty of permissions on its own. But the script then runs nothing like a deployed Lambda would: in production the agent runs under a narrow, single-purpose role, not your broad dev profile, and exercising that path now catches role-related issues before they hit deploy.

scripts/chat.py

def assume_role(role_arn: str) -> dict[str, str]:
    """Trade the local SSO identity for short-lived terraform-pr-agent creds."""
    sts = boto3.client("sts")
    response = sts.assume_role(RoleArn=role_arn, RoleSessionName="chat-cli")
    return response["Credentials"]

Build the model

Now we can put everything together and build the model. There is a small wrinkle here: we pass both the inference profile and the model ID to the model builder. The inference profile is what is actually called and the model ID is what helps pydantic-ai to use the right capabilities, but you could also overwrite them yourself in bedrock_additional_model_request_fields, which you may have to do for newer or less popular models where pydantic-ai's defaults are missing or broken. Some settings like thinking may also interfere with enforced tool calls.

scripts/chat.py

def build_model(role_arn: str, inference_profile_arn: str, region: str) -> BedrockConverseModel:
    creds = assume_role(role_arn)
    bedrock_client = boto3.client(
        "bedrock-runtime",
        region_name=region,
        aws_access_key_id=creds["AccessKeyId"],
        aws_secret_access_key=creds["SecretAccessKey"],
        aws_session_token=creds["SessionToken"],
    )
    # The application inference profile ARN goes in settings, not model_name:
    # pydantic-ai still uses the foundation-model id for capability detection
    # and routes the actual Converse call through the profile.
    return BedrockConverseModel(
        MODEL_ID,
        provider=BedrockProvider(bedrock_client=bedrock_client),
        settings={"bedrock_inference_profile": inference_profile_arn},
    )

The conversation loop

Finally, the agent loop, to chat with the agent. We will loop indefinitely until the agent provides us with the flag to terminate. We print whatever message the agent returns to the console. Give it a try!

⚠️ Warning

Anthropic models on Bedrock need special setup (see Anthropic First Time Use) - if you get errors go through the validation steps!

scripts/chat.py

    history: list = []
    while True:
        try:
            user_input = input("you> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if not user_input:
            continue
        try:
            result = agent.run_sync(user_input, message_history=history)
        except ClientError as err:
            sys.stderr.write(f"bedrock error: {err}\n")
            continue
        reply = result.output
        print(f"agent> {reply.message}")
        history = result.all_messages()
        if reply.terminate:
            break

Notice in the example chat below that the LLM appears to remember my name. The LLM has no memory of its own; the recall comes from the message_history that we pass in and never reset. The fact that we do not reset it does not matter for our toy example, but in production you would need to manage what gets passed as history and what the agent "forgets". Additionally, things like RAG (retrieval-augmented generation) could be used to give a longer-term memory (often semantic search with a vector store is used for this). It sounds complicated, but is often just enriching the user prompt (your first message to the agent) with semantically matching context from past conversations.

CloudWatch dashboard for inference profile metrics

If you tried out the agent in the previous section, you should have seen some metrics in CloudWatch. The screenshot below from my sandbox account shows a cryptic model ID, which is our inference profile ID. This is not very human-readable, but we'll fix that in this section with a CloudWatch dashboard that surfaces the metrics worth watching.

The dashboard scopes every widget to the terraform-pr-agent application inference profile via the ModelId dimension (Bedrock reuses that dimension name for the inference profile ID). Each metric carries a label override so the legend shows the friendly profile name rather than the ID or formula. Four widgets in a 2x2 grid:

Tokens: InputTokenCount and OutputTokenCount (sum). Your usage and cost driver.
Cache tokens: CacheReadInputTokenCount and CacheWriteInputTokenCount (sum). Stays at zero - Haiku usage in this post sits below Bedrock's caching threshold, and we have not opted into prompt caching yet anyway.
Invocations and errors: Invocations plus InvocationClientErrors, InvocationServerErrors, and InvocationThrottles (sum). Health at a glance.
Latency (ms): InvocationLatency average and p99.

Copy the file below and save it as infra/cloudwatch.tf in the project we started, then run terraform apply.

infra/cloudwatch.tf

locals {
  cloudwatch_region = data.aws_region.current.region

  dashboard_label = aws_bedrock_inference_profile.agent.name
  profile_id      = aws_bedrock_inference_profile.agent.id
}

resource "aws_cloudwatch_dashboard" "agent" {
  dashboard_name = aws_bedrock_inference_profile.agent.name
  dashboard_body = jsonencode({
    widgets = [
      {
        type   = "metric"
        x      = 0
        y      = 0
        width  = 12
        height = 6
        properties = {
          title  = "Tokens"
          region = local.cloudwatch_region
          view   = "timeSeries"
          stat   = "Sum"
          period = 60
          metrics = [
            # here we plug in the labels so this looks nicer
            ["AWS/Bedrock", "InputTokenCount", "ModelId", local.profile_id, { label = "${local.dashboard_label} / input" }],
            # the dot syntax allows not to repeat the same as above
            [".", "OutputTokenCount", ".", ".", { label = "${local.dashboard_label} / output" }],
          ]
        }
      },
      {
        type   = "metric"
        x      = 12
        y      = 0
        width  = 12
        height = 6
        properties = {
          title  = "Cache tokens"
          region = local.cloudwatch_region
          view   = "timeSeries"
          stat   = "Sum"
          period = 60
          metrics = [
            ["AWS/Bedrock", "CacheReadInputTokenCount", "ModelId", local.profile_id, { label = "${local.dashboard_label} / cache read" }],
            [".", "CacheWriteInputTokenCount", ".", ".", { label = "${local.dashboard_label} / cache write" }],
          ]
        }
      },
      {
        type   = "metric"
        x      = 0
        y      = 6
        width  = 12
        height = 6
        properties = {
          title  = "Invocations and errors"
          region = local.cloudwatch_region
          view   = "timeSeries"
          stat   = "Sum"
          period = 60
          metrics = [
            ["AWS/Bedrock", "Invocations", "ModelId", local.profile_id, { label = "${local.dashboard_label} / invocations" }],
            [".", "InvocationClientErrors", ".", ".", { label = "${local.dashboard_label} / 4xx" }],
            [".", "InvocationServerErrors", ".", ".", { label = "${local.dashboard_label} / 5xx" }],
            [".", "InvocationThrottles", ".", ".", { label = "${local.dashboard_label} / throttles" }],
          ]
        }
      },
      {
        type   = "metric"
        x      = 12
        y      = 6
        width  = 12
        height = 6
        properties = {
          title  = "Latency (ms)"
          region = local.cloudwatch_region
          view   = "timeSeries"
          period = 60
          metrics = [
            ["AWS/Bedrock", "InvocationLatency", "ModelId", local.profile_id, { label = "${local.dashboard_label} / avg", stat = "Average" }],
            [".", ".", ".", ".", { label = "${local.dashboard_label} / p99", stat = "p99" }],
          ]
        }
      },
    ]
  })
}

output "cloudwatch_dashboard_url" {
  value = format(
    "https://%s.console.aws.amazon.com/cloudwatch/home?region=%s#dashboards/dashboard/%s",
    local.cloudwatch_region,
    local.cloudwatch_region,
    aws_cloudwatch_dashboard.agent.dashboard_name,
  )
}

After terraform apply, the cloudwatch_dashboard_url output points straight at the new dashboard.

Daily token threshold alarm

You can copy the file below into infra/alerts.tf and run terraform apply to create an alarm that will yell when the token usage exceeds a daily threshold.

infra/alerts.tf

# alias/aws/sns is the AWS-managed SNS key; a CMK is unnecessary for an
# operational alarm topic that publishes alarm-fired events, not audit data.
#trivy:ignore:avd-aws-0136
resource "aws_sns_topic" "agent_alerts" {
  name              = "terraform-pr-agent-alerts"
  kms_master_key_id = "alias/aws/sns"
}

# The subscription stays PENDING until the recipient clicks the AWS
# confirmation email. Alarms fire either way, but no email goes out until the
# subscription is confirmed.
resource "aws_sns_topic_subscription" "agent_alerts_email" {
  topic_arn = aws_sns_topic.agent_alerts.arn
  protocol  = "email"
  endpoint  = var.alert_email
}

resource "aws_cloudwatch_metric_alarm" "daily_tokens" {
  alarm_name          = "${aws_bedrock_inference_profile.agent.name}-daily-tokens"
  alarm_description   = "Daily input + output token usage for the agent inference profile exceeded the configured threshold."
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  threshold           = var.daily_token_alarm_threshold
  treat_missing_data  = "notBreaching"

  metric_query {
    id          = "total"
    expression  = "input + output"
    label       = "Total tokens (input + output)"
    return_data = true
  }

  metric_query {
    id = "input"
    metric {
      namespace   = "AWS/Bedrock"
      metric_name = "InputTokenCount"
      dimensions  = { ModelId = local.profile_id }
      stat        = "Sum"
      period      = 86400
    }
  }

  metric_query {
    id = "output"
    metric {
      namespace   = "AWS/Bedrock"
      metric_name = "OutputTokenCount"
      dimensions  = { ModelId = local.profile_id }
      stat        = "Sum"
      period      = 86400
    }
  }

  alarm_actions = [aws_sns_topic.agent_alerts.arn]
}

Before terraform apply, set TF_VAR_alert_email in .envrc.local so the SNS subscription has somewhere to send. The block is already in the template, just uncomment it and run direnv reload:

.envrc.local

# Email subscribed to the alarms SNS topic. Required by infra/alerts.tf.
# Set before `terraform apply`; AWS sends a confirmation email that must be
# clicked before alarms can deliver.
# export TF_VAR_alert_email=you@example.com

This wires up a plain email SNS subscription. In production you would point the alarm at PagerDuty or incident.io so someone is actually paged. AWS sends a confirmation email on first apply, and the subscription stays PendingConfirmation until you click the link. The alarm itself fires regardless, but no email leaves SNS until that confirmation is in.

infra/alerts.tf

# alias/aws/sns is the AWS-managed SNS key; a CMK is unnecessary for an
# operational alarm topic that publishes alarm-fired events, not audit data.
#trivy:ignore:avd-aws-0136
resource "aws_sns_topic" "agent_alerts" {
  name              = "terraform-pr-agent-alerts"
  kms_master_key_id = "alias/aws/sns"
}

# The subscription stays PENDING until the recipient clicks the AWS
# confirmation email. Alarms fire either way, but no email goes out until the
# subscription is confirmed.
resource "aws_sns_topic_subscription" "agent_alerts_email" {
  topic_arn = aws_sns_topic.agent_alerts.arn
  protocol  = "email"
  endpoint  = var.alert_email
}

The alarm sums daily input and output tokens. Output tokens are more expensive, so you could weight them higher in the expression, or compute a dollar cost that also folds in cache read/write.

infra/alerts.tf

resource "aws_cloudwatch_metric_alarm" "daily_tokens" {
  alarm_name          = "${aws_bedrock_inference_profile.agent.name}-daily-tokens"
  alarm_description   = "Daily input + output token usage for the agent inference profile exceeded the configured threshold."
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  threshold           = var.daily_token_alarm_threshold
  treat_missing_data  = "notBreaching"

  metric_query {
    id          = "total"
    expression  = "input + output"
    label       = "Total tokens (input + output)"
    return_data = true
  }

  metric_query {
    id = "input"
    metric {
      namespace   = "AWS/Bedrock"
      metric_name = "InputTokenCount"
      dimensions  = { ModelId = local.profile_id }
      stat        = "Sum"
      period      = 86400
    }
  }

  metric_query {
    id = "output"
    metric {
      namespace   = "AWS/Bedrock"
      metric_name = "OutputTokenCount"
      dimensions  = { ModelId = local.profile_id }
      stat        = "Sum"
      period      = 86400
    }
  }

  alarm_actions = [aws_sns_topic.agent_alerts.arn]
}

End state

A pydantic-ai agent talking to Claude on Bedrock, a live dashboard showing token usage, and an alarm armed to yell when daily spend creeps up. Every later post in the series leans on this baseline.

Coming next: an audit trail for the agent.

DEV Community: Andreas Lang

Validators steer, evals grade: picking a model on cost per pass

What this post covers

From validators to evals

The result

The code

More validation tools

TaskResult: structured self-report

The eval harness: the validators are the conditions

Running the sweep

End state

File tools and validate: the agent's feedback loop

What this post covers

Architecture

The code

From zip to container image

ECR and the placeholder image

Watching cold starts

Workspace setup

File tools

The validate tool: agent-driven, any time

Caller-side retry

Parking the output in S3

Tracing the whole invocation

What validate catches and what it doesn't

End state

Model portability: swapping Bedrock for the Mistral API

What this post covers

Architecture

The model registry

Building the model at invoke time

Provider-agnostic metrics with EMF

End State

Audit trail via OTLP: every agent run as a trace

What this post covers

Architecture

OTLP schema for audit

Infra

Variables

The Lambda

KMS and the audit bucket

Firehose

Build and deploy

Wiring pydantic-ai to OTLP

Fetching the Logfire token at first invoke

The per-trace audit processor

Shipping one trace per Firehose record

Constructing the Bedrock model

The handler entry

Querying traces

DuckDB on your laptop

Athena/Snowflake

The same trace in Logfire

End state

Bedrock and pydantic-ai with observability from day one

Prerequisites

Tooling

AWS access

Anthropic First Time Use

Project scaffold

What this post covers

Bedrock IAM and inference profile

First pydantic-ai call

Add the new env vars to .envrc.local

Full script

System prompt - Basic Chat

Reply schema

Assume the role

Build the model

The conversation loop

CloudWatch dashboard for inference profile metrics

Daily token threshold alarm

End state

`TaskResult`: structured self-report

Add the new env vars to `.envrc.local`