OpsPilot Agent Core

Introduction

30 Second Summary

Employee departures can touch dozens of accounts before one person finishes their final day. A rushed handover can strand a device or leave access active too long.

In this project, you will build OpsPilot Agent Core, a serverless IT operations agent on AWS that uses Amazon Nova Micro through Amazon Bedrock. You will persist each offboarding workflow in Amazon DynamoDB so backend policy can pause a mutating action for authorized approval.

What You'll Build

From PowerShell, you will watch OpsPilot carry one offboarding request from evidence gathering to an approval pause before completing the simulated access change exactly once.

By the end of this project, you'll have:

• An LLM-directed tool loop that lets Amazon Nova Micro choose employee tools across multiple model turns.
• A durable approval checkpoint that keeps read-only tools autonomous while simulated access changes wait for the allowed role.
• Replay-safe outcomes backed by correlated JSON traces in Amazon CloudWatch Logs, giving you proof of model turns, policy decisions, approvals, token usage, and final results.
• Secret Mission: Add an incident-awareness tool so the LLM decides when open incident context matters before offboarding.

Are there any prerequisites?

You need an AWS account with an identity that can create AWS CloudFormation stacks, IAM roles, AWS Lambda functions, and Amazon DynamoDB tables. That identity also needs permission to create log groups in Amazon CloudWatch Logs plus budgets in AWS Budgets.

Before We Start

This first checkpoint locks in what you are building before any hands-on work begins. You are committing to a long-lived OpsPilot core where Amazon Nova Micro can choose useful tools while deterministic backend policy authorizes every mutating action.

Set Up Secure AWS Tooling

OpsPilot's backend controls matter only when you can deploy the same serverless stack from a trusted local session. Your Windows machine needs the AWS CLI before that deployment can be reproduced.

The AWS SAM CLI packages the serverless application for cloud deployment. Remote AWS testing lets Docker stay stopped for this project.

Short-lived AWS credentials connect PowerShell to us-east-1 without creating project files. The existing repository in GitHub remains the long-lived OpsPilot workspace in VS Code.

In this step, get ready to:

• Install AWS CLI 2.39.6 for your Windows user.
• Install AWS SAM CLI 1.171.0 for remote deployment.
• Connect the existing OpsPilot workspace to a verified short-lived AWS session.

Install AWS CLI 2

AWS CLI gives PowerShell a direct way to authenticate with AWS services. OpsPilot uses version 2.39.6 so every deployment starts from the same command-line environment.

• Press the Windows key to open Windows Search.
• Type PowerShell into the search field.
• Press Enter to open PowerShell.
• Check your current AWS CLI version by running this command:

aws --version

What Does This Command Check?

This command asks the installed AWS CLI to report its version. The result determines whether you can continue or need the current installer.

✔️ I see AWS CLI 2.39.6

Your output reports version 2.39.6. AWS CLI is ready for short-lived authentication.

ⓧ I see an older version

Your installed AWS CLI is older than the project version. The official PowerShell installer updates AWS CLI for your current Windows user.

• Install the current AWS CLI 2 release by running this command:

irm https://awscli.amazonaws.com/v2/install.ps1 | iex

What Does This Installer Do?

This command downloads the official AWS CLI 2 PowerShell installer. It installs the command for your current Windows user.

• Close the current PowerShell window.
• Open a new PowerShell window from Windows Search.
• Confirm the installed version by running this command:

aws --version

What Should This Confirm?

The new PowerShell window reloads the available commands. AWS CLI should now report version 2.39.6.

🙋‍♀️ AWS CLI Still Shows the Old Version?

Make sure you opened a new PowerShell window after the installer completed. A window that stayed open can keep the previous command path.

If the version still differs, help me troubleshoot the AWS CLI installation.

ⓧ Command not found

AWS CLI is not installed for your Windows user yet. The official PowerShell installer adds it without creating any OpsPilot resources.

• Install AWS CLI 2 for your current user by running this command:

irm https://awscli.amazonaws.com/v2/install.ps1 | iex

What Does This Installer Do?

This command downloads the official AWS CLI 2 PowerShell installer. It installs the command for your current Windows user.

• Close the current PowerShell window.
• Open a new PowerShell window from Windows Search.
• Confirm the installation by running this command:

aws --version

What Should This Confirm?

The new PowerShell window loads the installed command. AWS CLI should report version 2.39.6.

🙋‍♀️ AWS CLI Command Still Missing?

Check that the installer finished before you closed PowerShell. Open another new PowerShell window after the installation completes.

If the command remains unavailable, help me diagnose the missing AWS CLI command.

Install AWS SAM CLI

AWS SAM CLI turns a serverless template into a deployable AWS application. Version 1.171.0 can build this project without a running Docker engine.

• Check your current AWS SAM CLI version by running this command:

sam --version

What Does This Command Check?

This command reports the installed AWS SAM CLI version. The result shows whether the project can use its pinned deployment tool.

✔️ I see AWS SAM CLI 1.171.0

Your output reports version 1.171.0. AWS SAM CLI is ready to build the project in AWS.

ⓧ I see an older version

Your AWS SAM CLI version is older than the project version. The current Windows installer updates the tool.

• Visit the official AWS SAM CLI release page.

You should see release 1.171.0 on the page. Its Windows installer is named AWS_SAM_CLI_64_PY3.msi.

• Download AWS_SAM_CLI_64_PY3.msi from the release page.
• Run the downloaded installer.

The Windows installer opens after the download completes. Its default installation options provide the command required by this project.

• Complete the installation with the default options.
• Open a new PowerShell window from Windows Search.
• Confirm the updated version by running this command:

sam --version

What Should This Confirm?

The command should now report AWS SAM CLI version 1.171.0. This confirms the installer is available in the new PowerShell session.

🙋‍♀️ AWS SAM CLI Still Shows the Old Version?

Confirm that you downloaded AWS_SAM_CLI_64_PY3.msi from the 1.171.0 release. Open another new PowerShell window after completing the installer.

If the old version remains, help me troubleshoot the AWS SAM CLI update.

ⓧ Command not found

AWS SAM CLI is not installed yet. The release page provides the Windows installer required for this project.

• Visit the official AWS SAM CLI release page.

You should see release 1.171.0 on the page. Its Windows installer is named AWS_SAM_CLI_64_PY3.msi.

• Download AWS_SAM_CLI_64_PY3.msi from the release page.
• Run the downloaded installer.

The Windows installer opens after the download completes. Its default installation options add the AWS SAM CLI command.

• Complete the installation with the default options.
• Open a new PowerShell window from Windows Search.
• Confirm the installation by running this command:

sam --version

What Should This Confirm?

The command should report AWS SAM CLI version 1.171.0. This confirms PowerShell can find the installed tool.

🙋‍♀️ AWS SAM CLI Command Still Missing?

Confirm that the installer completed before opening the new PowerShell window. Make sure the downloaded file was AWS_SAM_CLI_64_PY3.msi.

If the command remains unavailable, help me diagnose the AWS SAM CLI installation.

That is the local deployment toolkit sorted. Both AWS command-line tools now report the project versions.

Connect the OpsPilot Workspace

Short-lived console credentials give AWS CLI a temporary session in us-east-1. This keeps long-lived access keys out of the OpsPilot repository.

The login command starts a browser sign-in using your eligible AWS console identity. Complete that browser flow before returning to PowerShell.

• Close the PowerShell window used for installation.
• Press the Windows key to open Windows Search.
• Type PowerShell into the search field.
• Press Enter to open a new PowerShell window.
• Start the short-lived AWS login in us-east-1 by running this command:

aws login --region us-east-1

What Does This Login Do?

AWS CLI uses your browser to establish temporary console credentials. The Region argument prepares the session for OpsPilot resources in us-east-1.

• Complete the browser sign-in with your eligible AWS console identity.

Before you check the session, what identity details do you expect AWS to return?

• Verify the active AWS identity by running this command:

aws sts get-caller-identity

What Does This Identity Check Prove?

• The Account value identifies the AWS account used by your session.
• The Arn value identifies the active AWS principal.
• A successful response proves that PowerShell can make authenticated AWS requests.

You should see your AWS Account value in the response. You should also see the active Arn.

Identity Check Did Not Succeed?

Confirm that the browser sign-in finished before you ran the identity check. Repeat the short-lived login command above if the session did not complete.

If AWS still cannot return your identity, help me troubleshoot the short-lived AWS session.

Opening the existing repository as a VS Code workspace gives the editor access to its project context. Workspace Trust may appear when the folder opens.

Trust Only Your Repository

Workspace Trust controls whether code from the opened folder can run in VS Code. Full trust is appropriate only when you recognize the GitHub-backed OpsPilot repository.

• Press the Windows key to open Windows Search.
• Type VS Code into the search field.
• Press Enter to open VS Code.
• Press Ctrl+K Ctrl+O to open the folder picker.
• Select your existing GitHub-backed OpsPilot repository folder.
• Click Select Folder.
• Review the folder source if Workspace Trust appears.
• Select Yes, I trust the authors only for your own repository.

You should see the OpsPilot repository folder at the top of the Explorer view. This confirms that VS Code opened the long-lived workspace.

Before you run the final check, which four details do you expect PowerShell to confirm?

• Switch back to the new PowerShell window from earlier.
• Verify the complete toolchain plus the active AWS session by running these commands:

aws --version
sam --version
python --version
aws sts get-caller-identity

What Does This Final Check Cover?

• The first command checks the AWS CLI installation.
• The second command checks the AWS SAM CLI installation.
• The third command checks the existing Python installation.
• The final command checks the active short-lived AWS identity.

You should see AWS CLI version 2.39.6. You should also see AWS SAM CLI version 1.171.0.

Python should report version 3.13.14. The identity response should contain your AWS Account.

The identity response should also contain your active Arn. Together these results prove that the secure deployment environment is ready.

One of the Final Checks Failed?

Open a fresh PowerShell window if either installed command is missing. Repeat the matching installation path above if a version differs.

Repeat the browser login if the identity command fails. For a result that still differs, help me review the complete Windows toolchain.

Your secure Windows toolchain now points at a verified AWS identity. Next, you will deploy OpsPilot's first Bedrock-backed response without relying on Docker.

Deploy the First Bedrock Decision

Your secure AWS session is ready inside the long-lived OpsPilot repository. You can now turn that empty workspace into a deployed agent core.

A plain model response from Amazon Bedrock gives you the first visible proof that OpsPilot can reach an LLM. This text-only version also reveals the boundary you will address next because it cannot inspect enterprise state or take controlled action.

In this step, get ready to:

• Create the initial Lambda source plus its deployment infrastructure.
• Build the application before deploying its CloudFormation stack.
• Invoke Amazon Nova Micro through Bedrock from PowerShell.

Create the serverless foundation

The AWS Serverless Application Model describes your application as infrastructure as code. AWS CloudFormation uses that description to create the same resources whenever you deploy OpsPilot.

What will this foundation create?

• An AWS Lambda function runs the initial model request with a concurrency limit of two.
• An Amazon DynamoDB table prepares the project for persistent workflow state in the next iteration.
• Amazon CloudWatch Logs stores JSON logs for three days.
• AWS Budgets sends an account-wide alert when actual monthly spend passes half of the $2 budget.

• Create `src/requirements.txt` inside the repository by using the VS Code file sidebar.

You should see an empty `requirements.txt` tab inside the new `src` folder.

• Pin the two Python dependencies by pasting these lines into `src/requirements.txt`:

boto3==1.43.126
botocore==1.43.126

What do these dependencies provide?

• `boto3` gives the Lambda function a Python client for Amazon Bedrock.
• `botocore` supplies the lower-level AWS request support used by Boto3.

• Save `src/requirements.txt`.

The dependency file now contains two pinned packages that AWS SAM can install consistently.

• Create `src/app.py` inside the repository by using the VS Code file sidebar.

You should see an empty `app.py` tab beside `requirements.txt`.

• Add the Bedrock client plus the response text helper by pasting this first chunk into `src/app.py`:

import os

import boto3

MODEL_ID = os.environ.get("MODEL_ID", "amazon.nova-micro-v1:0")
bedrock = boto3.client("bedrock-runtime")


def extract_text(message):
    return " ".join(
        block["text"]
        for block in message.get("content", [])
        if "text" in block
    ).strip()

What does this code do?

• `MODEL_ID` reads the deployed model setting while keeping `amazon.nova-micro-v1:0` as its fallback.
• `bedrock` creates the Bedrock Runtime client used for model requests.
• `extract_text()` joins the text blocks returned by Amazon Nova Micro into one response string.

• Save `src/app.py`.

The top of `app.py` now shows the model configuration plus the `extract_text()` helper.

Seeing red underlines in app.py?

Check that each opening parenthesis has a matching closing parenthesis. Keep the indentation inside `extract_text()` exactly as shown.

Ask for help with the first source chunk: help me compare my Bedrock client and extract_text helper with the required Python structure

• Add request validation plus the first model handler below `extract_text()` by pasting this chunk:

def validation_error(detail):
    return {
        "ok": False,
        "status": "validation_error",
        "error": {"type": "ValidationError", "detail": detail}
    }


def handle_ask(event):
    request = str(event.get("request", "")).strip()
    if not request:
        return validation_error("request is required")
    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=[{"role": "user", "content": [{"text": request}]}],
        inferenceConfig={"maxTokens": 160, "temperature": 0}
    )
    message = response["output"]["message"]
    return {
        "ok": True,
        "status": "completed",
        "model_id": MODEL_ID,
        "response": extract_text(message)
    }

How does the first decision work?

• `validation_error()` produces a structured response when the event has no request text.
• `handle_ask()` sends the request to Bedrock Converse with a maximum of 160 output tokens.
• A temperature of `0` keeps tool-capable Nova requests deterministic enough for later orchestration work.
• The successful response exposes its status plus the model ID and generated text.

• Save `src/app.py`.

The source file now contains the complete `handle_ask()` path for a text-only model response.

Does the handler look misaligned?

Make sure every line inside `handle_ask()` uses four spaces of indentation. Confirm that `inferenceConfig` remains inside the `bedrock.converse()` call.

Ask for help with the handler structure: help me find an indentation or structure problem in my handle_ask function

• Add the Lambda entry point below `handle_ask()` by pasting this final chunk:

def lambda_handler(event, context):
    if not isinstance(event, dict):
        return validation_error("event must be a JSON object")

    action = event.get("action")
    if action == "ask":
        return handle_ask(event)

What does the entry point do?

• `lambda_handler()` rejects an event that is not a JSON object.
• The `ask` action routes directly to `handle_ask()`.
• No tool-selection loop exists in this first version.

• Save `src/app.py`.

The initial Lambda source now has one runnable path from `lambda_handler()` to Amazon Bedrock.

Is lambda_handler missing?

Confirm that `lambda_handler()` starts at the left edge of the file. Check that the call uses the exact `handle_ask` identifier defined above it.

Ask for help with the entry point: help me verify that my Lambda entry point routes ask events correctly

• Create `events/ask.json` inside the repository by using the VS Code file sidebar.

You should see an empty `ask.json` tab inside the new `events` folder.

• Define the first model request by pasting this event into `events/ask.json`:

{"action":"ask","request":"In one sentence, explain why enterprise AI agents need deterministic approval boundaries."}

What does this event test?

The `action` value selects `handle_ask()`. The `request` value asks the model for a concise explanation that you can verify in the remote response.

• Save `events/ask.json`.

The repository now has a repeatable event for testing the same model path after every deployment.

• Create `template.yaml` at the top level of the repository by using the VS Code file sidebar.

You should see an empty `template.yaml` tab beside the `src` and `events` folders.

• Add the template header plus the workflow table by pasting this first chunk into `template.yaml`:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: OpsPilot Agent Core

Parameters:
  BudgetEmail:
    Type: String
    Description: Email address for the account-wide OpsPilot cost alert

Resources:
  WorkflowTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: workflow_id
          AttributeType: S
      KeySchema:
        - AttributeName: workflow_id
          KeyType: HASH
      Tags:
        - Key: Project
          Value: OpsPilot

What does this template section define?

• `BudgetEmail` accepts the address that receives the account-wide cost alert.
• `WorkflowTable` uses on-demand billing through `PAY_PER_REQUEST`.
• `workflow_id` becomes the string partition key for future saved workflows.

• Save `template.yaml`.

The template now contains its deployment parameter plus the durable state resource.

• Add the Lambda runtime configuration below the `WorkflowTable` resource by pasting this chunk:

  AgentFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: opspilot-agent-core
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.13
      Architectures:
        - x86_64
      MemorySize: 256
      Timeout: 120
      ReservedConcurrentExecutions: 2
      LoggingConfig:
        LogFormat: JSON
        ApplicationLogLevel: INFO
        SystemLogLevel: WARN
      Environment:
        Variables:
          WORKFLOW_TABLE: !Ref WorkflowTable
          MODEL_ID: amazon.nova-micro-v1:0
          MAX_AGENT_STEPS: '6'
          ENFORCE_APPROVAL: 'true'

How is the Lambda bounded?

• The function runs on the Python `3.13` runtime with 256 MB of memory.
• A 120-second timeout limits each invocation.
• Reserved concurrency limits simultaneous executions to two.
• The environment fixes the model plus the future loop limit and approval setting.

• Save `template.yaml`.

The template now connects `app.lambda_handler` to a bounded Lambda configuration.

• Add the Lambda permissions directly below the `Environment` section by pasting this chunk:

      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref WorkflowTable
        - Version: '2012-10-17'
          Statement:
            - Sid: InvokeNovaMicro
              Effect: Allow
              Action:
                - bedrock:InvokeModel
              Resource: !Sub arn:${AWS::Partition}:bedrock:${AWS::Region}::foundation-model/amazon.nova-micro-v1:0
      Tags:
        Project: OpsPilot

Why are these permissions scoped?

The DynamoDB policy targets the table created by this template. The Bedrock statement permits only model invocation against the Nova Micro foundation-model ARN in the deployment Region.

• Save `template.yaml`.

The Lambda resource now has the permissions required for this deployment without a broad Bedrock wildcard.

• Add the log group plus the monthly budget below the `AgentFunction` resource by pasting this chunk:

  AgentLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: /aws/lambda/opspilot-agent-core
      RetentionInDays: 3

  CostBudget:
    Type: AWS::Budgets::Budget
    Properties:
      Budget:
        BudgetName: opspilot-agent-core-monthly
        BudgetLimit:
          Amount: 2
          Unit: USD
        TimeUnit: MONTHLY
        BudgetType: COST
      NotificationsWithSubscribers:
        - Notification:
            NotificationType: ACTUAL
            ComparisonOperator: GREATER_THAN
            Threshold: 50
            ThresholdType: PERCENTAGE
          Subscribers:
            - SubscriptionType: EMAIL
              Address: !Ref BudgetEmail

How do the operational guardrails work?

• `AgentLogGroup` keeps Lambda logs for three days.
• `CostBudget` monitors actual account-wide monthly cost against a $2 amount.
• The email notification triggers when actual spending passes 50 percent of that amount.

• Save `template.yaml`.

The stack now includes short log retention plus an account-wide cost signal.

• Add the stack outputs below the budget resource by pasting this final chunk:

Outputs:
  FunctionName:
    Value: !Ref AgentFunction
  WorkflowTableName:
    Value: !Ref WorkflowTable
  LogGroupName:
    Value: !Ref AgentLogGroup

What do the outputs expose?

The outputs reveal the deployed function name plus the workflow table name and log group name. These values make the stack easier to inspect after deployment.

• Save `template.yaml`.

The repository now contains every source file required for the first deployment.

Does the YAML indentation look uneven?

Use spaces throughout `template.yaml`. Keep each resource aligned beneath `Resources` while keeping `Outputs` at the left edge.

Ask for help with the template structure: help me find an indentation problem in my SAM template

Deploy the stack with AWS SAM

The build stage validates the project shape before AWS receives anything. The guided deployment then records your choices in `samconfig.toml` for later redeployments.

• Switch back to the PowerShell window from the previous step.
• Build the application from the open OpsPilot repository by running this command:

sam build

What does this command do?

`sam build` reads `template.yaml` plus the Python dependency file. It prepares the deployable application without using Docker.

You should see a successful build summary that points to the built template.

Did the build fail?

Check that PowerShell is running from the folder containing `template.yaml`. Confirm that `src/requirements.txt` uses the two exact dependency pins.

Ask for help with the build output: help me diagnose my failed OpsPilot sam build

This deployment creates usage-based AWS resources. The $2 budget provides an account-wide alert while your application remains deliberately small.

The guided deployment is interactive. Expect the stack creation phase to take a few minutes while CloudFormation provisions each resource.

• Enter `opspilot-agent-core` when the deployment asks for the stack name.
• Enter `us-east-1` when the deployment asks for the Region.
• Enter your budget email when the deployment asks for `BudgetEmail`.
• Allow the deployment to create the IAM role.
• Confirm the proposed change set.
• Save the deployment settings to `samconfig.toml`.
• Start the guided deployment by running this command:

sam deploy --guided

What does guided deployment save?

AWS SAM deploys the stack through CloudFormation. It also writes the stack name plus the Region and parameter choices into `samconfig.toml`.

You should see the deployment finish with the `opspilot-agent-core` stack in a completed state. The output also lists the function plus the table and log group.

Did the deployment stop?

Confirm that your short-lived AWS session is still active. Check that your identity can create CloudFormation stacks plus IAM roles and Lambda functions.

Ask for help with the deployment output: help me troubleshoot my failed OpsPilot SAM guided deployment

✔️ Awesome, I've got everything!

Your source files are saved. Your generated `samconfig.toml` records the guided deployment settings for `opspilot-agent-core` in `us-east-1`.

ⓧ I'd like to double check the full code

`src/requirements.txt`:

boto3==1.43.126
botocore==1.43.126

`src/app.py`:

import os

import boto3

MODEL_ID = os.environ.get("MODEL_ID", "amazon.nova-micro-v1:0")
bedrock = boto3.client("bedrock-runtime")


def extract_text(message):
    return " ".join(
        block["text"]
        for block in message.get("content", [])
        if "text" in block
    ).strip()


def validation_error(detail):
    return {
        "ok": False,
        "status": "validation_error",
        "error": {"type": "ValidationError", "detail": detail}
    }


def handle_ask(event):
    request = str(event.get("request", "")).strip()
    if not request:
        return validation_error("request is required")
    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=[{"role": "user", "content": [{"text": request}]}],
        inferenceConfig={"maxTokens": 160, "temperature": 0}
    )
    message = response["output"]["message"]
    return {
        "ok": True,
        "status": "completed",
        "model_id": MODEL_ID,
        "response": extract_text(message)
    }


def lambda_handler(event, context):
    if not isinstance(event, dict):
        return validation_error("event must be a JSON object")

    action = event.get("action")
    if action == "ask":
        return handle_ask(event)

`events/ask.json`:

{"action":"ask","request":"In one sentence, explain why enterprise AI agents need deterministic approval boundaries."}

`template.yaml`:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: OpsPilot Agent Core

Parameters:
  BudgetEmail:
    Type: String
    Description: Email address for the account-wide OpsPilot cost alert

Resources:
  WorkflowTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: workflow_id
          AttributeType: S
      KeySchema:
        - AttributeName: workflow_id
          KeyType: HASH
      Tags:
        - Key: Project
          Value: OpsPilot

  AgentFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: opspilot-agent-core
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.13
      Architectures:
        - x86_64
      MemorySize: 256
      Timeout: 120
      ReservedConcurrentExecutions: 2
      LoggingConfig:
        LogFormat: JSON
        ApplicationLogLevel: INFO
        SystemLogLevel: WARN
      Environment:
        Variables:
          WORKFLOW_TABLE: !Ref WorkflowTable
          MODEL_ID: amazon.nova-micro-v1:0
          MAX_AGENT_STEPS: '6'
          ENFORCE_APPROVAL: 'true'
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref WorkflowTable
        - Version: '2012-10-17'
          Statement:
            - Sid: InvokeNovaMicro
              Effect: Allow
              Action:
                - bedrock:InvokeModel
              Resource: !Sub arn:${AWS::Partition}:bedrock:${AWS::Region}::foundation-model/amazon.nova-micro-v1:0
      Tags:
        Project: OpsPilot

  AgentLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: /aws/lambda/opspilot-agent-core
      RetentionInDays: 3

  CostBudget:
    Type: AWS::Budgets::Budget
    Properties:
      Budget:
        BudgetName: opspilot-agent-core-monthly
        BudgetLimit:
          Amount: 2
          Unit: USD
        TimeUnit: MONTHLY
        BudgetType: COST
      NotificationsWithSubscribers:
        - Notification:
            NotificationType: ACTUAL
            ComparisonOperator: GREATER_THAN
            Threshold: 50
            ThresholdType: PERCENTAGE
          Subscribers:
            - SubscriptionType: EMAIL
              Address: !Ref BudgetEmail

Outputs:
  FunctionName:
    Value: !Ref AgentFunction
  WorkflowTableName:
    Value: !Ref WorkflowTable
  LogGroupName:
    Value: !Ref AgentLogGroup

The generated `samconfig.toml` stores the stack name `opspilot-agent-core` plus the Region `us-east-1`. It also records IAM role creation and the supplied `BudgetEmail` parameter.

Invoke the first Bedrock decision

Remote invocation sends `events/ask.json` to the deployed `AgentFunction`. The Lambda then calls Bedrock Converse and returns the model text to PowerShell.

Before you run the check, do you expect the response to include enterprise records or only a generated explanation?

• Invoke the deployed ask event by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/ask.json --region us-east-1

What does this command do?

• `AgentFunction` identifies the logical Lambda resource in the SAM stack.
• `events/ask.json` supplies the request payload.
• `us-east-1` targets the Region that contains the deployed stack and Nova Micro access.

You should see `status` set to `completed`. The response also names `amazon.nova-micro-v1:0` and includes a short Bedrock-generated answer.

That answer is the first working OpsPilot decision. Its text-only result is the intended limitation because no enterprise tool supplied evidence or performed an action.

Did the remote invocation fail?

Confirm that the stack name is `opspilot-agent-core`. Check that your short-lived AWS session remains active in `us-east-1`.

If the Lambda returns a failed response, inspect the error detail without changing the model ID or event shape.

Ask for help with the invocation response: help me diagnose my failed remote invocation of AgentFunction

You have a live Bedrock-powered Lambda with reproducible infrastructure plus cost and logging guardrails. Next, you will give the LLM dynamic enterprise tools so its decisions can gather evidence and request actions.

Give the LLM Dynamic Tools

Your deployed diagnostic endpoint proves that Amazon Bedrock can generate a response with Nova Micro. The function still depends on the request to contain every piece of enterprise context.

An LLM becomes an agent when it can choose tools based on the goal. In this step, you will add a client-side tool loop that stores each turn in Amazon DynamoDB before executing a simulated access change.

In this step, get ready to:

• Register three simulated enterprise tools with automatic tool selection.
• Expand the Lambda into a bounded tool-use loop with persistent workflow state.
• Run a simulated offboarding workflow that exposes unguarded mutation.

Register the enterprise tools

A tool registry describes the operations available to the model. Your Python code remains responsible for dispatching the operation that the model selects.

Why client-side tool use?

Client-side tool use keeps the planner-executor loop visible in your application code. You can inspect each tool request before your backend executes it.

This approach keeps persistence and policy decisions inside the agent core. Those boundaries remain under your control.

• Switch back to the OpsPilot workspace in VS Code.
• Create src/tools.py inside the existing src folder from the Explorer sidebar.
• Use the full-code comparison tab below to add the tool registry to src/tools.py.

✔️ Awesome, I've got everything!

Your tool registry contains three schemas with matching Python handlers. Save src/tools.py before continuing.

ⓧ I'd like to double check the full code

import hashlib
import json

MUTATING_TOOLS = {"disable_employee_access"}

TOOL_CONFIG = {
    "tools": [
        {
            "toolSpec": {
                "name": "get_employee_profile",
                "description": "Read an employee profile before deciding how to handle an IT operations request.",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {
                            "employee_id": {"type": "string"}
                        },
                        "required": ["employee_id"]
                    }
                }
            }
        },
        {
            "toolSpec": {
                "name": "get_assigned_assets",
                "description": "Read the systems and devices currently assigned to an employee.",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {
                            "employee_id": {"type": "string"}
                        },
                        "required": ["employee_id"]
                    }
                }
            }
        },
        {
            "toolSpec": {
                "name": "disable_employee_access",
                "description": "Disable an employee's access to specified systems. This is a mutating action.",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {
                            "employee_id": {"type": "string"},
                            "systems": {
                                "type": "array",
                                "items": {"type": "string"}
                            },
                            "reason": {"type": "string"}
                        },
                        "required": ["employee_id", "systems", "reason"]
                    }
                }
            }
        }
    ],
    "toolChoice": {"auto": {}}
}

EMPLOYEES = {
    "EMP-1042": {
        "employee_id": "EMP-1042",
        "name": "Morgan Lee",
        "department": "Finance Operations",
        "manager": "Priya Shah",
        "employment_status": "departing"
    }
}

ASSETS = {
    "EMP-1042": {
        "employee_id": "EMP-1042",
        "systems": ["Google Workspace", "GitHub Enterprise", "Jira"],
        "devices": ["Windows laptop NW-4821"],
        "open_asset_actions": ["Return laptop NW-4821"]
    }
}


def get_employee_profile(employee_id):
    return EMPLOYEES.get(
        employee_id,
        {"employee_id": employee_id, "found": False}
    )


def get_assigned_assets(employee_id):
    return ASSETS.get(
        employee_id,
        {"employee_id": employee_id, "systems": [], "devices": []}
    )


def disable_employee_access(employee_id, systems, reason):
    receipt_source = json.dumps(
        {
            "employee_id": employee_id,
            "systems": systems,
            "reason": reason
        },
        sort_keys=True
    )
    receipt = hashlib.sha256(receipt_source.encode("utf-8")).hexdigest()[:12]
    return {
        "executed": True,
        "mode": "simulation",
        "employee_id": employee_id,
        "systems_disabled": systems,
        "reason": reason,
        "execution_id": f"sim-{receipt}"
    }


TOOL_HANDLERS = {
    "get_employee_profile": get_employee_profile,
    "get_assigned_assets": get_assigned_assets,
    "disable_employee_access": disable_employee_access
}


def execute_tool(tool_name, tool_input):
    handler = TOOL_HANDLERS.get(tool_name)
    if handler is None:
        raise ValueError(f"Unknown tool: {tool_name}")
    return handler(**tool_input)

How the registry works

• The TOOL_CONFIG schemas tell Nova Micro which inputs each tool accepts.
• The auto tool choice lets the model select the next useful operation.
• The sample records give the read-only tools deterministic enterprise data.
• The TOOL_HANDLERS map dispatches each selected name to its Python implementation.

• Save src/tools.py.
• Confirm that src/tools.py appears under the src folder in the Explorer sidebar.

Tool file missing or incomplete?

Check that the file is inside the existing src folder. Confirm that its name ends with .py.

Compare the opening imports with the final execute_tool() function. Both must be in the same file.

Help me check my tool registry.

Build the persistent tool loop

The registry describes individual operations. The AWS Lambda function now needs a bounded planner-executor loop that returns each tool result to the same model conversation.

Each workflow serializes its messages and trace as JSON in DynamoDB. A deterministic workflow ID lets repeated request IDs resolve to the same record.

• Return to the existing src/app.py tab.
• Replace its current text-only Lambda code with the complete agent host from the full-code comparison tab below.

✔️ Awesome, I've got everything!

Your Lambda host now imports the tool registry. It also persists workflow messages and structured traces.

ⓧ I'd like to double check the full code

import hashlib
import json
import logging
import os
from datetime import datetime, timezone

import boto3
from botocore.exceptions import ClientError

from tools import MUTATING_TOOLS, TOOL_CONFIG, execute_tool

MODEL_ID = os.environ.get("MODEL_ID", "amazon.nova-micro-v1:0")
TABLE_NAME = os.environ.get("WORKFLOW_TABLE", "")
MAX_AGENT_STEPS = int(os.environ.get("MAX_AGENT_STEPS", "6"))
ENFORCE_APPROVAL = os.environ.get("ENFORCE_APPROVAL", "true").lower() == "true"

bedrock = boto3.client("bedrock-runtime")
dynamodb = boto3.resource("dynamodb")
logger = logging.getLogger()
logger.setLevel(logging.INFO)

SYSTEM_PROMPT = """You are OpsPilot, an enterprise IT operations agent.
Use the available tools for operational requests instead of inventing enterprise data.
Choose one tool at a time. Gather relevant read-only evidence before requesting a mutating tool.
Never claim a tool executed unless its tool result says it executed.
When the request is complete, provide a concise outcome summary."""


def now_iso():
    return datetime.now(timezone.utc).isoformat()


def encode(value):
    return json.dumps(value, separators=(",", ":"), sort_keys=True)


def decode(item, field, default):
    value = item.get(field)
    return json.loads(value) if value else default


def stable_hash(value):
    return hashlib.sha256(encode(value).encode("utf-8")).hexdigest()


def workflow_id_for(request_id):
    digest = hashlib.sha256(request_id.encode("utf-8")).hexdigest()[:16]
    return f"wf-{digest}"


def workflow_table():
    if not TABLE_NAME:
        raise RuntimeError("WORKFLOW_TABLE is not configured")
    return dynamodb.Table(TABLE_NAME)


def log_event(event_type, **fields):
    logger.info(
        event_type,
        extra={
            "event_type": event_type,
            "agent_id": "opspilot-agent-core",
            **fields
        }
    )


def add_trace(item, event_type, **data):
    trace = decode(item, "trace_json", [])
    entry = {"timestamp": now_iso(), "event_type": event_type, **data}
    trace.append(entry)
    item["trace_json"] = encode(trace)
    log_event(
        event_type,
        trace_id=item.get("workflow_id", "none"),
        workflow_id=item.get("workflow_id", "none"),
        **data
    )


def persist(item):
    item["updated_at"] = now_iso()
    workflow_table().put_item(Item=item)


def load_workflow(request_id):
    workflow_id = workflow_id_for(request_id)
    response = workflow_table().get_item(Key={"workflow_id": workflow_id})
    return response.get("Item")


def public_view(item, duplicate=False):
    pending = decode(item, "pending_action_json", None)
    result = decode(item, "action_result_json", None)
    return {
        "ok": item.get("status") != "failed",
        "workflow_id": item["workflow_id"],
        "request_id": item["request_id"],
        "status": item["status"],
        "model_id": MODEL_ID,
        "duplicate": duplicate,
        "pending_action": pending,
        "action_result": result,
        "final_response": item.get("final_response"),
        "error": decode(item, "error_json", None),
        "trace": decode(item, "trace_json", [])
    }


def validation_error(detail):
    return {
        "ok": False,
        "status": "validation_error",
        "error": {"type": "ValidationError", "detail": detail}
    }


def extract_text(message):
    return " ".join(
        block["text"]
        for block in message.get("content", [])
        if "text" in block
    ).strip()


def fail_workflow(item, error):
    item["status"] = "failed"
    item["error_json"] = encode(
        {"type": type(error).__name__, "detail": str(error)[:300]}
    )
    add_trace(
        item,
        "workflow_failed",
        error_type=type(error).__name__,
        outcome="failed"
    )
    persist(item)
    return public_view(item)


def run_agent(item):
    messages = decode(item, "messages_json", [])
    first_iteration = int(item.get("step_count", 0))

    for iteration in range(first_iteration, MAX_AGENT_STEPS):
        response = bedrock.converse(
            modelId=MODEL_ID,
            system=[{"text": SYSTEM_PROMPT}],
            messages=messages,
            toolConfig=TOOL_CONFIG,
            inferenceConfig={"maxTokens": 500, "temperature": 0}
        )

        output_message = response["output"]["message"]
        stop_reason = response["stopReason"]
        usage = response.get("usage", {})
        messages.append(output_message)
        item["messages_json"] = encode(messages)
        item["step_count"] = iteration + 1

        add_trace(
            item,
            "model_turn",
            model_id=MODEL_ID,
            iteration=iteration + 1,
            stop_reason=stop_reason,
            input_tokens=usage.get("inputTokens", 0),
            output_tokens=usage.get("outputTokens", 0),
            outcome="received"
        )

        if stop_reason == "end_turn":
            item["status"] = "completed"
            item["final_response"] = extract_text(output_message)
            add_trace(item, "workflow_completed", outcome="completed")
            persist(item)
            return public_view(item)

        if stop_reason != "tool_use":
            raise RuntimeError(f"Unexpected Bedrock stop reason: {stop_reason}")

        tool_requests = [
            block["toolUse"]
            for block in output_message.get("content", [])
            if "toolUse" in block
        ]
        if len(tool_requests) != 1:
            raise RuntimeError("OpsPilot requires exactly one tool request per model turn")

        tool_request = tool_requests[0]
        tool_name = tool_request["name"]
        tool_input = tool_request["input"]
        add_trace(
            item,
            "tool_requested",
            iteration=iteration + 1,
            tool_name=tool_name,
            tool_input=tool_input,
            outcome="requested"
        )

        if ENFORCE_APPROVAL and tool_name in MUTATING_TOOLS:
            item["status"] = "awaiting_approval"
            item["pending_action_json"] = encode(tool_request)
            add_trace(
                item,
                "approval_required",
                tool_name=tool_name,
                policy="mutating_tools_require_ops_approver",
                outcome="blocked"
            )
            persist(item)
            return public_view(item)

        tool_result = execute_tool(tool_name, tool_input)
        messages.append(
            {
                "role": "user",
                "content": [
                    {
                        "toolResult": {
                            "toolUseId": tool_request["toolUseId"],
                            "content": [{"json": tool_result}]
                        }
                    }
                ]
            }
        )
        item["messages_json"] = encode(messages)
        if tool_name in MUTATING_TOOLS:
            item["action_result_json"] = encode(tool_result)
        add_trace(
            item,
            "tool_executed",
            tool_name=tool_name,
            execution_id=tool_result.get("execution_id", "read-only"),
            outcome="executed"
        )
        persist(item)

    raise RuntimeError("Agent exceeded MAX_AGENT_STEPS")


def handle_ask(event):
    request = str(event.get("request", "")).strip()
    if not request:
        return validation_error("request is required")
    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=[{"role": "user", "content": [{"text": request}]}],
        inferenceConfig={"maxTokens": 160, "temperature": 0}
    )
    message = response["output"]["message"]
    return {
        "ok": True,
        "status": "completed",
        "model_id": MODEL_ID,
        "response": extract_text(message)
    }


def handle_start(event):
    request_id = str(event.get("request_id", "")).strip()
    request = str(event.get("request", "")).strip()
    requester = event.get("requester", {})
    if not request_id or not request or not isinstance(requester, dict):
        return validation_error("request_id, request, and requester are required")

    workflow_id = workflow_id_for(request_id)
    payload_hash = stable_hash({"request": request, "requester": requester})
    item = {
        "workflow_id": workflow_id,
        "request_id": request_id,
        "payload_hash": payload_hash,
        "requester_json": encode(requester),
        "status": "running",
        "step_count": 0,
        "created_at": now_iso(),
        "updated_at": now_iso(),
        "messages_json": encode(
            [{"role": "user", "content": [{"text": request}]}]
        ),
        "trace_json": encode([])
    }
    add_trace(item, "workflow_created", outcome="created")

    try:
        workflow_table().put_item(
            Item=item,
            ConditionExpression="attribute_not_exists(workflow_id)"
        )
    except ClientError as error:
        if error.response["Error"]["Code"] != "ConditionalCheckFailedException":
            raise
        existing = load_workflow(request_id)
        if existing["payload_hash"] != payload_hash:
            return {
                "ok": False,
                "status": "conflict",
                "workflow_id": existing["workflow_id"],
                "error": {
                    "type": "IdempotencyConflict",
                    "detail": "request_id was reused with a different payload"
                }
            }
        return public_view(existing, duplicate=True)

    try:
        return run_agent(item)
    except Exception as error:
        return fail_workflow(item, error)


def handle_status(event):
    request_id = str(event.get("request_id", "")).strip()
    if not request_id:
        return validation_error("request_id is required")
    item = load_workflow(request_id)
    if item is None:
        return {"ok": False, "status": "not_found", "request_id": request_id}
    return public_view(item)


def handle_approve(event):
    request_id = str(event.get("request_id", "")).strip()
    actor = event.get("actor", {})
    if not request_id or not isinstance(actor, dict):
        return validation_error("request_id and actor are required")

    workflow_id = workflow_id_for(request_id)
    if actor.get("role") != "ops_approver":
        log_event(
            "authorization_denied",
            trace_id=workflow_id,
            workflow_id=workflow_id,
            actor_id=actor.get("id", "unknown"),
            actor_role=actor.get("role", "unknown"),
            outcome="forbidden"
        )
        return {
            "ok": False,
            "status": "forbidden",
            "workflow_id": workflow_id,
            "error": {
                "type": "AuthorizationError",
                "detail": "ops_approver role is required"
            }
        }

    item = load_workflow(request_id)
    if item is None:
        return {"ok": False, "status": "not_found", "request_id": request_id}
    if item["status"] == "completed":
        return public_view(item, duplicate=True)
    if item["status"] != "awaiting_approval":
        return {
            "ok": False,
            "status": item["status"],
            "workflow_id": workflow_id,
            "error": {
                "type": "StateConflict",
                "detail": "workflow is not awaiting approval"
            }
        }

    try:
        response = workflow_table().update_item(
            Key={"workflow_id": workflow_id},
            UpdateExpression=(
                "SET #status = :executing, approved_by = :actor, "
                "updated_at = :updated_at"
            ),
            ConditionExpression="#status = :awaiting",
            ExpressionAttributeNames={"#status": "status"},
            ExpressionAttributeValues={
                ":executing": "executing",
                ":awaiting": "awaiting_approval",
                ":actor": actor,
                ":updated_at": now_iso()
            },
            ReturnValues="ALL_NEW"
        )
        item = response["Attributes"]
    except ClientError as error:
        if error.response["Error"]["Code"] != "ConditionalCheckFailedException":
            raise
        return public_view(load_workflow(request_id), duplicate=True)

    pending = decode(item, "pending_action_json", None)
    try:
        tool_result = execute_tool(pending["name"], pending["input"])
        messages = decode(item, "messages_json", [])
        messages.append(
            {
                "role": "user",
                "content": [
                    {
                        "toolResult": {
                            "toolUseId": pending["toolUseId"],
                            "content": [{"json": tool_result}]
                        }
                    }
                ]
            }
        )
        item["messages_json"] = encode(messages)
        item["action_result_json"] = encode(tool_result)
        item["status"] = "running"
        item.pop("pending_action_json", None)
        add_trace(
            item,
            "approval_granted",
            actor_id=actor.get("id", "unknown"),
            actor_role=actor["role"],
            tool_name=pending["name"],
            outcome="approved"
        )
        add_trace(
            item,
            "tool_executed",
            tool_name=pending["name"],
            execution_id=tool_result["execution_id"],
            outcome="executed"
        )
        persist(item)
        return run_agent(item)
    except Exception as error:
        return fail_workflow(item, error)


def lambda_handler(event, context):
    if not isinstance(event, dict):
        return validation_error("event must be a JSON object")

    action = event.get("action")
    try:
        if action == "ask":
            return handle_ask(event)
        if action == "start":
            return handle_start(event)
        if action == "status":
            return handle_status(event)
        if action == "approve":
            return handle_approve(event)
        return validation_error("action must be ask, start, status, or approve")
    except Exception as error:
        logger.exception(
            "unhandled_error",
            extra={
                "event_type": "unhandled_error",
                "agent_id": "opspilot-agent-core",
                "action": str(action),
                "error_type": type(error).__name__
            }
        )
        return {
            "ok": False,
            "status": "failed",
            "error": {"type": type(error).__name__, "detail": str(error)[:300]}
        }

What does the agent host do?

• The hashing helpers derive a stable workflow ID from each request ID.
• The persistence helpers serialize conversation messages and trace entries into the workflow table.
• The run_agent() loop sends the current messages and tool schemas to Bedrock.
• Each toolResult message gives the model verified backend output for its next turn.
• The iteration limit prevents an open-ended sequence of model calls.

• Save src/app.py.
• Build the expanded application from the existing repository by running:

sam build

What does this build check?

AWS SAM prepares the Lambda source with the dependencies from src/requirements.txt. A successful build confirms that the application structure is ready for deployment.

You should see a successful build message. The expanded Lambda package now includes app.py plus tools.py.

Build did not complete?

Confirm that app.py plus tools.py are inside the existing src folder. Check that the import uses the exact module name tools.

Compare the indentation around run_agent() with the full-code tab. Python treats indentation as part of the program structure.

Help me diagnose my SAM build.

Run the unguarded workflow

The new agent host can process a multi-turn request. You will now deploy an intentionally incomplete policy state before checking its behavior.

• Return to the existing template.yaml tab.
• Find this environment variable:

          ENFORCE_APPROVAL: 'true'

What does the current value control?

The environment variable controls whether run_agent() pauses a tool listed in MUTATING_TOOLS.

• Replace the current line with this value:

          ENFORCE_APPROVAL: 'false'

What changes in this deployment?

The value false sends every selected tool to execute_tool() without applying the mutating-tool pause.

• Save template.yaml.
• Create events/start-unsafe.json inside the existing events folder from the Explorer sidebar.
• Add the simulated offboarding request with this code:

{
  "action": "start",
  "request_id": "demo-offboarding-unsafe-001",
  "request": "Offboard employee EMP-1042. Review relevant records and disable access to the systems you determine are assigned. The final working day is today.",
  "requester": {
    "id": "recruiter-demo",
    "role": "ops_requester"
  }
}

What does this event request?

The start action creates a persistent workflow for demo-offboarding-unsafe-001. Its request gives the model an operational goal without prescribing a fixed tool sequence.

The employee data and asset data come from the backend tools. The model decides which evidence it needs.

• Save events/start-unsafe.json.
• Rebuild the application with the updated configuration by running:

sam build

Why build again?

The new build captures the updated Lambda source plus deployment configuration. This keeps the packaged application aligned with the repository.

You should see another successful build message for the updated application.

• Redeploy the existing opspilot-agent-core stack by running:

sam deploy

What does this deployment update?

SAM uses the saved settings in samconfig.toml. AWS CloudFormation updates the Lambda code plus its environment configuration.

This update can take several minutes while AWS replaces the function package. The PowerShell prompt returns when the stack update finishes.

Deployment failed?

Confirm that your short-lived AWS session is still active. The deployment also needs the Region plus stack settings stored in samconfig.toml.

Review the failed stack event shown by SAM. It identifies the resource that blocked the update.

Help me troubleshoot this deployment.

Before you invoke the workflow, what do you think happens when the model requests an access-changing tool while the approval switch is disabled?

• Invoke the unsafe offboarding event against the deployed Lambda by running:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/start-unsafe.json --region us-east-1

What does this invocation test?

The command sends events/start-unsafe.json to the deployed AgentFunction. The Lambda keeps calling Nova Micro until the model finishes or the loop reaches its step limit.

You should see status set to completed. The response also names amazon.nova-micro-v1:0 as the model.

The simulated disable_employee_access tool executed without approval. That is the deliberate shortfall this run was designed to expose.

The unsafe result is the lesson

The trace reaches tool_executed for the mutating tool. Its deterministic execution_id begins with sim-.

Prompt instructions influenced the model's choices. They did not create an independent authorization boundary.

🙋‍♀️ Workflow failed or skipped the tool?

Confirm that the deployed template sets ENFORCE_APPROVAL to false. Confirm that the event uses the request ID from the full-code tab.

A repeated invocation returns the workflow already stored for that request ID. Use the first-run response when checking the model's tool sequence.

Help me inspect the workflow response.

✔️ Awesome, I've got everything!

Your repository now contains the tool registry plus unsafe start event. The deployed stack has approval enforcement disabled for this demonstration.

ⓧ I'd like to double check the full code

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: OpsPilot Agent Core

Parameters:
  BudgetEmail:
    Type: String
    Description: Email address for the account-wide OpsPilot cost alert

Resources:
  WorkflowTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: workflow_id
          AttributeType: S
      KeySchema:
        - AttributeName: workflow_id
          KeyType: HASH
      Tags:
        - Key: Project
          Value: OpsPilot

  AgentFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: opspilot-agent-core
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.13
      Architectures:
        - x86_64
      MemorySize: 256
      Timeout: 120
      ReservedConcurrentExecutions: 2
      LoggingConfig:
        LogFormat: JSON
        ApplicationLogLevel: INFO
        SystemLogLevel: WARN
      Environment:
        Variables:
          WORKFLOW_TABLE: !Ref WorkflowTable
          MODEL_ID: amazon.nova-micro-v1:0
          MAX_AGENT_STEPS: '6'
          ENFORCE_APPROVAL: 'false'
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref WorkflowTable
        - Version: '2012-10-17'
          Statement:
            - Sid: InvokeNovaMicro
              Effect: Allow
              Action:
                - bedrock:InvokeModel
              Resource: !Sub arn:${AWS::Partition}:bedrock:${AWS::Region}::foundation-model/amazon.nova-micro-v1:0
      Tags:
        Project: OpsPilot

  AgentLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: /aws/lambda/opspilot-agent-core
      RetentionInDays: 3

  CostBudget:
    Type: AWS::Budgets::Budget
    Properties:
      Budget:
        BudgetName: opspilot-agent-core-monthly
        BudgetLimit:
          Amount: 2
          Unit: USD
        TimeUnit: MONTHLY
        BudgetType: COST
      NotificationsWithSubscribers:
        - Notification:
            NotificationType: ACTUAL
            ComparisonOperator: GREATER_THAN
            Threshold: 50
            ThresholdType: PERCENTAGE
          Subscribers:
            - SubscriptionType: EMAIL
              Address: !Ref BudgetEmail

Outputs:
  FunctionName:
    Value: !Ref AgentFunction
  WorkflowTableName:
    Value: !Ref WorkflowTable
  LogGroupName:
    Value: !Ref AgentLogGroup

What should you compare?

Confirm that the workflow table plus existing cost controls remain unchanged. The environment now sets ENFORCE_APPROVAL to false for the deliberate demonstration.

{
  "action": "start",
  "request_id": "demo-offboarding-unsafe-001",
  "request": "Offboard employee EMP-1042. Review relevant records and disable access to the systems you determine are assigned. The final working day is today.",
  "requester": {
    "id": "recruiter-demo",
    "role": "ops_requester"
  }
}

What should the event contain?

The event uses the start action plus the stable unsafe demonstration request ID. The requester role remains ops_requester.

You now have a persistent agent loop that lets Nova Micro choose enterprise tools. Next, you will turn the unsafe result into a durable approval boundary.

Enforce a Durable Approval Gate

You proved that your Amazon Bedrock tool loop can gather evidence and complete an offboarding request. You also saw the simulated access change execute without approval.

That result exposes the boundary this step must enforce. Backend policy must pause a mutating tool and preserve its model conversation in DynamoDB until an authorized role claims the action.

In this step, get ready to:

• Activate approval enforcement for mutating tool requests.
• Confirm the role check and conditional approval claim.
• Prove that a paused workflow survives across separate Lambda invocations.

Activate the backend approval boundary

The approval code already sits inside your bounded agent loop. Its behavior remained dormant during the unsafe demonstration because ENFORCE_APPROVAL was set to false in the deployed environment.

• Switch back to src/app.py in VS Code.
• Press Cmd+F (macOS) or Ctrl+F (Windows) to open the editor search.
• Find if ENFORCE_APPROVAL and tool_name in MUTATING_TOOLS:.
• Confirm that the branch sets status to awaiting_approval.
• Confirm that the branch stores the model request in pending_action_json.
• Confirm that the branch records an approval_required trace before returning.

Why does this gate hold?

The model can request disable_employee_access. Only application code can decide whether that request executes.

Read-only tools stay autonomous because they are absent from MUTATING_TOOLS. The policy boundary targets the action that changes access.

• Find def handle_approve(event): in src/app.py.
• Confirm that actor.get("role") must equal ops_approver.
• Find the update_item() call inside handle_approve().
• Confirm that its condition only changes awaiting_approval to executing.

What does the conditional update protect?

The conditional update is a compare-and-set transition. Only one approval attempt can claim a workflow while its saved status is awaiting_approval.

A competing attempt receives the stored workflow instead of running the pending tool again. This creates the foundation for replay-safe approval.

✔️ Awesome, I've got everything!

Your existing src/app.py contains the complete approval boundary. Continue to the deployment setting.

ⓧ I'd like to double check the full code

Compare your existing src/app.py with this complete baseline.

import hashlib
import json
import logging
import os
from datetime import datetime, timezone

import boto3
from botocore.exceptions import ClientError

from tools import MUTATING_TOOLS, TOOL_CONFIG, execute_tool

MODEL_ID = os.environ.get("MODEL_ID", "amazon.nova-micro-v1:0")
TABLE_NAME = os.environ.get("WORKFLOW_TABLE", "")
MAX_AGENT_STEPS = int(os.environ.get("MAX_AGENT_STEPS", "6"))
ENFORCE_APPROVAL = os.environ.get("ENFORCE_APPROVAL", "true").lower() == "true"

bedrock = boto3.client("bedrock-runtime")
dynamodb = boto3.resource("dynamodb")
logger = logging.getLogger()
logger.setLevel(logging.INFO)

SYSTEM_PROMPT = """You are OpsPilot, an enterprise IT operations agent.
Use the available tools for operational requests instead of inventing enterprise data.
Choose one tool at a time. Gather relevant read-only evidence before requesting a mutating tool.
Never claim a tool executed unless its tool result says it executed.
When the request is complete, provide a concise outcome summary."""


def now_iso():
    return datetime.now(timezone.utc).isoformat()


def encode(value):
    return json.dumps(value, separators=(",", ":"), sort_keys=True)


def decode(item, field, default):
    value = item.get(field)
    return json.loads(value) if value else default


def stable_hash(value):
    return hashlib.sha256(encode(value).encode("utf-8")).hexdigest()


def workflow_id_for(request_id):
    digest = hashlib.sha256(request_id.encode("utf-8")).hexdigest()[:16]
    return f"wf-{digest}"


def workflow_table():
    if not TABLE_NAME:
        raise RuntimeError("WORKFLOW_TABLE is not configured")
    return dynamodb.Table(TABLE_NAME)


def log_event(event_type, **fields):
    logger.info(
        event_type,
        extra={
            "event_type": event_type,
            "agent_id": "opspilot-agent-core",
            **fields
        }
    )


def add_trace(item, event_type, **data):
    trace = decode(item, "trace_json", [])
    entry = {"timestamp": now_iso(), "event_type": event_type, **data}
    trace.append(entry)
    item["trace_json"] = encode(trace)
    log_event(
        event_type,
        trace_id=item.get("workflow_id", "none"),
        workflow_id=item.get("workflow_id", "none"),
        **data
    )


def persist(item):
    item["updated_at"] = now_iso()
    workflow_table().put_item(Item=item)


def load_workflow(request_id):
    workflow_id = workflow_id_for(request_id)
    response = workflow_table().get_item(Key={"workflow_id": workflow_id})
    return response.get("Item")


def public_view(item, duplicate=False):
    pending = decode(item, "pending_action_json", None)
    result = decode(item, "action_result_json", None)
    return {
        "ok": item.get("status") != "failed",
        "workflow_id": item["workflow_id"],
        "request_id": item["request_id"],
        "status": item["status"],
        "model_id": MODEL_ID,
        "duplicate": duplicate,
        "pending_action": pending,
        "action_result": result,
        "final_response": item.get("final_response"),
        "error": decode(item, "error_json", None),
        "trace": decode(item, "trace_json", [])
    }


def validation_error(detail):
    return {
        "ok": False,
        "status": "validation_error",
        "error": {"type": "ValidationError", "detail": detail}
    }


def extract_text(message):
    return " ".join(
        block["text"]
        for block in message.get("content", [])
        if "text" in block
    ).strip()


def fail_workflow(item, error):
    item["status"] = "failed"
    item["error_json"] = encode(
        {"type": type(error).__name__, "detail": str(error)[:300]}
    )
    add_trace(
        item,
        "workflow_failed",
        error_type=type(error).__name__,
        outcome="failed"
    )
    persist(item)
    return public_view(item)


def run_agent(item):
    messages = decode(item, "messages_json", [])
    first_iteration = int(item.get("step_count", 0))

    for iteration in range(first_iteration, MAX_AGENT_STEPS):
        response = bedrock.converse(
            modelId=MODEL_ID,
            system=[{"text": SYSTEM_PROMPT}],
            messages=messages,
            toolConfig=TOOL_CONFIG,
            inferenceConfig={"maxTokens": 500, "temperature": 0}
        )

        output_message = response["output"]["message"]
        stop_reason = response["stopReason"]
        usage = response.get("usage", {})
        messages.append(output_message)
        item["messages_json"] = encode(messages)
        item["step_count"] = iteration + 1

        add_trace(
            item,
            "model_turn",
            model_id=MODEL_ID,
            iteration=iteration + 1,
            stop_reason=stop_reason,
            input_tokens=usage.get("inputTokens", 0),
            output_tokens=usage.get("outputTokens", 0),
            outcome="received"
        )

        if stop_reason == "end_turn":
            item["status"] = "completed"
            item["final_response"] = extract_text(output_message)
            add_trace(item, "workflow_completed", outcome="completed")
            persist(item)
            return public_view(item)

        if stop_reason != "tool_use":
            raise RuntimeError(f"Unexpected Bedrock stop reason: {stop_reason}")

        tool_requests = [
            block["toolUse"]
            for block in output_message.get("content", [])
            if "toolUse" in block
        ]
        if len(tool_requests) != 1:
            raise RuntimeError("OpsPilot requires exactly one tool request per model turn")

        tool_request = tool_requests[0]
        tool_name = tool_request["name"]
        tool_input = tool_request["input"]
        add_trace(
            item,
            "tool_requested",
            iteration=iteration + 1,
            tool_name=tool_name,
            tool_input=tool_input,
            outcome="requested"
        )

        if ENFORCE_APPROVAL and tool_name in MUTATING_TOOLS:
            item["status"] = "awaiting_approval"
            item["pending_action_json"] = encode(tool_request)
            add_trace(
                item,
                "approval_required",
                tool_name=tool_name,
                policy="mutating_tools_require_ops_approver",
                outcome="blocked"
            )
            persist(item)
            return public_view(item)

        tool_result = execute_tool(tool_name, tool_input)
        messages.append(
            {
                "role": "user",
                "content": [
                    {
                        "toolResult": {
                            "toolUseId": tool_request["toolUseId"],
                            "content": [{"json": tool_result}]
                        }
                    }
                ]
            }
        )
        item["messages_json"] = encode(messages)
        if tool_name in MUTATING_TOOLS:
            item["action_result_json"] = encode(tool_result)
        add_trace(
            item,
            "tool_executed",
            tool_name=tool_name,
            execution_id=tool_result.get("execution_id", "read-only"),
            outcome="executed"
        )
        persist(item)

    raise RuntimeError("Agent exceeded MAX_AGENT_STEPS")


def handle_ask(event):
    request = str(event.get("request", "")).strip()
    if not request:
        return validation_error("request is required")
    response = bedrock.converse(
        modelId=MODEL_ID,
        messages=[{"role": "user", "content": [{"text": request}]}],
        inferenceConfig={"maxTokens": 160, "temperature": 0}
    )
    message = response["output"]["message"]
    return {
        "ok": True,
        "status": "completed",
        "model_id": MODEL_ID,
        "response": extract_text(message)
    }


def handle_start(event):
    request_id = str(event.get("request_id", "")).strip()
    request = str(event.get("request", "")).strip()
    requester = event.get("requester", {})
    if not request_id or not request or not isinstance(requester, dict):
        return validation_error("request_id, request, and requester are required")

    workflow_id = workflow_id_for(request_id)
    payload_hash = stable_hash({"request": request, "requester": requester})
    item = {
        "workflow_id": workflow_id,
        "request_id": request_id,
        "payload_hash": payload_hash,
        "requester_json": encode(requester),
        "status": "running",
        "step_count": 0,
        "created_at": now_iso(),
        "updated_at": now_iso(),
        "messages_json": encode(
            [{"role": "user", "content": [{"text": request}]}]
        ),
        "trace_json": encode([])
    }
    add_trace(item, "workflow_created", outcome="created")

    try:
        workflow_table().put_item(
            Item=item,
            ConditionExpression="attribute_not_exists(workflow_id)"
        )
    except ClientError as error:
        if error.response["Error"]["Code"] != "ConditionalCheckFailedException":
            raise
        existing = load_workflow(request_id)
        if existing["payload_hash"] != payload_hash:
            return {
                "ok": False,
                "status": "conflict",
                "workflow_id": existing["workflow_id"],
                "error": {
                    "type": "IdempotencyConflict",
                    "detail": "request_id was reused with a different payload"
                }
            }
        return public_view(existing, duplicate=True)

    try:
        return run_agent(item)
    except Exception as error:
        return fail_workflow(item, error)


def handle_status(event):
    request_id = str(event.get("request_id", "")).strip()
    if not request_id:
        return validation_error("request_id is required")
    item = load_workflow(request_id)
    if item is None:
        return {"ok": False, "status": "not_found", "request_id": request_id}
    return public_view(item)


def handle_approve(event):
    request_id = str(event.get("request_id", "")).strip()
    actor = event.get("actor", {})
    if not request_id or not isinstance(actor, dict):
        return validation_error("request_id and actor are required")

    workflow_id = workflow_id_for(request_id)
    if actor.get("role") != "ops_approver":
        log_event(
            "authorization_denied",
            trace_id=workflow_id,
            workflow_id=workflow_id,
            actor_id=actor.get("id", "unknown"),
            actor_role=actor.get("role", "unknown"),
            outcome="forbidden"
        )
        return {
            "ok": False,
            "status": "forbidden",
            "workflow_id": workflow_id,
            "error": {
                "type": "AuthorizationError",
                "detail": "ops_approver role is required"
            }
        }

    item = load_workflow(request_id)
    if item is None:
        return {"ok": False, "status": "not_found", "request_id": request_id}
    if item["status"] == "completed":
        return public_view(item, duplicate=True)
    if item["status"] != "awaiting_approval":
        return {
            "ok": False,
            "status": item["status"],
            "workflow_id": workflow_id,
            "error": {
                "type": "StateConflict",
                "detail": "workflow is not awaiting approval"
            }
        }

    try:
        response = workflow_table().update_item(
            Key={"workflow_id": workflow_id},
            UpdateExpression=(
                "SET #status = :executing, approved_by = :actor, "
                "updated_at = :updated_at"
            ),
            ConditionExpression="#status = :awaiting",
            ExpressionAttributeNames={"#status": "status"},
            ExpressionAttributeValues={
                ":executing": "executing",
                ":awaiting": "awaiting_approval",
                ":actor": actor,
                ":updated_at": now_iso()
            },
            ReturnValues="ALL_NEW"
        )
        item = response["Attributes"]
    except ClientError as error:
        if error.response["Error"]["Code"] != "ConditionalCheckFailedException":
            raise
        return public_view(load_workflow(request_id), duplicate=True)

    pending = decode(item, "pending_action_json", None)
    try:
        tool_result = execute_tool(pending["name"], pending["input"])
        messages = decode(item, "messages_json", [])
        messages.append(
            {
                "role": "user",
                "content": [
                    {
                        "toolResult": {
                            "toolUseId": pending["toolUseId"],
                            "content": [{"json": tool_result}]
                        }
                    }
                ]
            }
        )
        item["messages_json"] = encode(messages)
        item["action_result_json"] = encode(tool_result)
        item["status"] = "running"
        item.pop("pending_action_json", None)
        add_trace(
            item,
            "approval_granted",
            actor_id=actor.get("id", "unknown"),
            actor_role=actor["role"],
            tool_name=pending["name"],
            outcome="approved"
        )
        add_trace(
            item,
            "tool_executed",
            tool_name=pending["name"],
            execution_id=tool_result["execution_id"],
            outcome="executed"
        )
        persist(item)
        return run_agent(item)
    except Exception as error:
        return fail_workflow(item, error)


def lambda_handler(event, context):
    if not isinstance(event, dict):
        return validation_error("event must be a JSON object")

    action = event.get("action")
    try:
        if action == "ask":
            return handle_ask(event)
        if action == "start":
            return handle_start(event)
        if action == "status":
            return handle_status(event)
        if action == "approve":
            return handle_approve(event)
        return validation_error("action must be ask, start, status, or approve")
    except Exception as error:
        logger.exception(
            "unhandled_error",
            extra={
                "event_type": "unhandled_error",
                "agent_id": "opspilot-agent-core",
                "action": str(action),
                "error_type": type(error).__name__
            }
        )
        return {
            "ok": False,
            "status": "failed",
            "error": {"type": type(error).__name__, "detail": str(error)[:300]}
        }

What should this file contain?

This baseline includes the approval pause in run_agent() and the conditional claim in handle_approve(). It also preserves the existing model loop and workflow storage.

The runtime switch lives in your infrastructure template. Changing it activates the dormant branch for every new invocation after deployment.

• Switch back to template.yaml in VS Code.
• Find the ENFORCE_APPROVAL environment variable under AgentFunction.
• Replace its current value with 'true'.

✔️ Awesome, I've got everything!

Your template now activates the approval branch. Save the file before building.

ⓧ I'd like to double check the full code

Compare template.yaml with this complete baseline template.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: OpsPilot Agent Core

Parameters:
  BudgetEmail:
    Type: String
    Description: Email address for the account-wide OpsPilot cost alert

Resources:
  WorkflowTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: workflow_id
          AttributeType: S
      KeySchema:
        - AttributeName: workflow_id
          KeyType: HASH
      Tags:
        - Key: Project
          Value: OpsPilot

  AgentFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: opspilot-agent-core
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.13
      Architectures:
        - x86_64
      MemorySize: 256
      Timeout: 120
      ReservedConcurrentExecutions: 2
      LoggingConfig:
        LogFormat: JSON
        ApplicationLogLevel: INFO
        SystemLogLevel: WARN
      Environment:
        Variables:
          WORKFLOW_TABLE: !Ref WorkflowTable
          MODEL_ID: amazon.nova-micro-v1:0
          MAX_AGENT_STEPS: '6'
          ENFORCE_APPROVAL: 'true'
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref WorkflowTable
        - Version: '2012-10-17'
          Statement:
            - Sid: InvokeNovaMicro
              Effect: Allow
              Action:
                - bedrock:InvokeModel
              Resource: !Sub arn:${AWS::Partition}:bedrock:${AWS::Region}::foundation-model/amazon.nova-micro-v1:0
      Tags:
        Project: OpsPilot

  AgentLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: /aws/lambda/opspilot-agent-core
      RetentionInDays: 3

  CostBudget:
    Type: AWS::Budgets::Budget
    Properties:
      Budget:
        BudgetName: opspilot-agent-core-monthly
        BudgetLimit:
          Amount: 2
          Unit: USD
        TimeUnit: MONTHLY
        BudgetType: COST
      NotificationsWithSubscribers:
        - Notification:
            NotificationType: ACTUAL
            ComparisonOperator: GREATER_THAN
            Threshold: 50
            ThresholdType: PERCENTAGE
          Subscribers:
            - SubscriptionType: EMAIL
              Address: !Ref BudgetEmail

Outputs:
  FunctionName:
    Value: !Ref AgentFunction
  WorkflowTableName:
    Value: !Ref WorkflowTable
  LogGroupName:
    Value: !Ref AgentLogGroup

What should this template contain?

The template keeps the existing table and cost controls. Its ENFORCE_APPROVAL value now activates the mutating-tool gate.

• Save template.yaml.
• Build the updated application by running this command in the PowerShell terminal from earlier:

sam build

What does this command do?

The build prepares your current src code and updated template.yaml for deployment. It does not require Docker for this project.

You should see the build complete with deployment artifacts prepared from the updated template.

Build did not complete?

• Confirm that your PowerShell terminal is inside the existing OpsPilot repository.
• Check that template.yaml is saved with consistent YAML indentation.

Ask for help with an AWS SAM build failure after enabling the approval environment variable.

The deployment can take a few minutes while AWS updates the Lambda configuration. Your saved samconfig.toml supplies the stack name and Region.

• Deploy the approval-enabled application by running this command:

sam deploy

What does this command do?

The deployment uses the settings saved during your guided deployment. CloudFormation updates opspilot-agent-core so new Lambda invocations receive approval enforcement.

You should see the existing stack finish its update without creating a second stack.

Deployment update failed?

• Confirm that your short-lived AWS session is still active.
• Check that samconfig.toml still names opspilot-agent-core in us-east-1.

Ask for help with an AWS SAM deployment update failure for the OpsPilot stack.

Create the safe workflow events

The safe request uses a new deterministic request ID. Separate event files let you start the workflow and retrieve its persisted state through the same Lambda contract.

• Select the events folder in the VS Code Explorer sidebar.
• Use the file creation control to create start.json inside events.
• Paste this safe offboarding request into events/start.json:

{
  "action": "start",
  "request_id": "demo-offboarding-001",
  "request": "Offboard employee EMP-1042. Review relevant records and disable access to the systems you determine are assigned. The final working day is today.",
  "requester": {
    "id": "recruiter-demo",
    "role": "ops_requester"
  }
}

What does this event do?

The event creates the workflow for demo-offboarding-001 using an operations requester. That role can submit work without approving the eventual access change.

• Save events/start.json.
• Confirm that start.json appears under the events folder in the Explorer sidebar.

Cannot save the start event?

• Confirm that start.json is inside the existing events folder.
• Check that every JSON property uses double quotes.

Ask for help with checking the structure of the safe start event.

• Use the file creation control to create status.json inside the same events folder.
• Paste this status request into events/status.json:

{
  "action": "status",
  "request_id": "demo-offboarding-001"
}

What does this event do?

The status event asks handle_status() to load the item derived from the same request ID. It proves that the pause survives beyond the original Lambda process.

• Save events/status.json.
• Confirm that status.json appears beside start.json in the Explorer sidebar.

Status event is not listed?

• Expand the events folder in the Explorer sidebar.
• Confirm that the filename ends in .json.

Ask for help with finding the status event in the OpsPilot repository.

Two approval events preserve the actor roles used by the deterministic authorization check. You create them now so the next safety test can compare a requester with an approver.

• Use the file creation control to create approve-denied.json inside the events folder.
• Paste this requester approval attempt into events/approve-denied.json:

{
  "action": "approve",
  "request_id": "demo-offboarding-001",
  "actor": {
    "id": "casey.requester",
    "role": "ops_requester"
  }
}

What does this event represent?

This event carries an actor whose role is ops_requester. The backend role check treats that actor as unauthorized to approve a mutating action.

• Save events/approve-denied.json.
• Confirm that approve-denied.json appears in the Explorer sidebar.

Denied event looks malformed?

• Check that actor contains both id and role.
• Confirm that the role value remains ops_requester.

Ask for help with checking the denied approval event.

• Use the file creation control to create approve.json inside the events folder.
• Paste this authorized approval attempt into events/approve.json:

{
  "action": "approve",
  "request_id": "demo-offboarding-001",
  "actor": {
    "id": "alex.approver",
    "role": "ops_approver"
  }
}

What does this event represent?

This event carries the required ops_approver role. The conditional update can claim the pending workflow only after this deterministic role check passes.

• Save events/approve.json.
• Confirm that all four new event files appear under the events folder.

Approval event is missing?

• Confirm that approve.json is saved beside the other event files.
• Check that its role value remains ops_approver.

Ask for help with checking the authorized approval event.

✔️ Awesome, I've got everything!

Your safe request and approval event files are saved. The workflow is ready for its persisted pause test.

ⓧ I'd like to double check the full code

Compare each new event file with its complete version below.

{
  "action": "start",
  "request_id": "demo-offboarding-001",
  "request": "Offboard employee EMP-1042. Review relevant records and disable access to the systems you determine are assigned. The final working day is today.",
  "requester": {
    "id": "recruiter-demo",
    "role": "ops_requester"
  }
}

What should this file contain?

The start event contains the safe request ID and the requester context used to create the persisted workflow.

{
  "action": "status",
  "request_id": "demo-offboarding-001"
}

What should this file contain?

The status event contains the same request ID so the Lambda can retrieve the existing workflow.

{
  "action": "approve",
  "request_id": "demo-offboarding-001",
  "actor": {
    "id": "casey.requester",
    "role": "ops_requester"
  }
}

What should this file contain?

The denied event contains an actor with the requester role.

{
  "action": "approve",
  "request_id": "demo-offboarding-001",
  "actor": {
    "id": "alex.approver",
    "role": "ops_approver"
  }
}

What should this file contain?

The authorized event contains an actor with the approver role.

Prove the workflow pauses durably

The next invocation reaches the same model-directed tool path as the unsafe demonstration. The active policy now intercepts the mutating request before execute_tool() can run it.

Before you run the safe request, do you expect its final status to show completion or a pause?

• Submit the safe offboarding request by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/start.json --region us-east-1

What does this command do?

The command invokes the deployed AgentFunction with events/start.json. The Lambda stores each model turn and tool result under the deterministic workflow ID.

You should see status set to awaiting_approval. The response should name disable_employee_access as the pending action and contain no execution receipt.

That pause is the intended result. Read-only tools ran autonomously while the mutating request stopped at the backend policy boundary.

Workflow completed instead of pausing?

• Confirm that the deployed template.yaml value is ENFORCE_APPROVAL: 'true'.
• Confirm that disable_employee_access remains in MUTATING_TOOLS.
• Rebuild the application after correcting either value.

Ask for help with a workflow that still executes the mutating tool without approval.

A separate status invocation now tests persistence. It loads the saved item without relying on memory from the Lambda process that started the workflow.

Before you run the status check, do you expect the pending tool request to remain available?

• Retrieve the persisted workflow by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/status.json --region us-east-1

What does this command prove?

The command starts a separate Lambda invocation and calls handle_status(). That handler derives the same wf- workflow ID and reads the persisted item from DynamoDB.

You should again see awaiting_approval with the pending disable_employee_access request. The trace should include approval_required without a mutating-tool execution receipt.

That is the hard part secured. Your agent can pause one model conversation and recover its exact approval state in a later invocation.

Status request cannot find the workflow?

• Confirm that events/start.json completed before you ran the status request.
• Check that both event files use demo-offboarding-001 as the request ID.
• Confirm that both commands target opspilot-agent-core in us-east-1.

Ask for help with a persisted workflow that cannot be retrieved by its status event.

Your model-directed workflow now stops at a durable authorization boundary. Next, you will challenge that boundary with denied requests, valid approval, retries, and structured logs.

Prove Safety and Idempotency

Your workflow now pauses before disable_employee_access can execute. The saved Amazon Bedrock conversation remains available in Amazon DynamoDB.

A production-oriented agent needs evidence that unauthorized callers cannot create side effects. Retries must also return the saved result.

Malformed requests need a handled response. Structured traces tie that safety evidence together.

In this step, get ready to:

• Reject an unauthorized approval.
• Complete the workflow exactly once.
• Verify replay safety through correlated traces.

Test denied and allowed approvals

The backend role check forms a deterministic authorization boundary. The model cannot override the required ops_approver value.

Before you run the denied approval, do you expect the requester role to pass the backend check? Hold that prediction.

• Invoke the denied approval event by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/approve-denied.json --region us-east-1

What Does This Invocation Prove?

The event identifies casey.requester with the ops_requester role. The backend rejects that role before claiming the pending action.

You should see status set to forbidden. The workflow remains paused because the mutating tool did not execute.

Now predict whether the allowed approval resumes the saved conversation or creates a fresh workflow. Hold that prediction.

• Invoke the allowed approval event by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/approve.json --region us-east-1

What Does the Allowed Approval Do?

The ops_approver role passes the authorization check. A conditional update claims the workflow only while its status is awaiting_approval.

The saved tool request executes after that claim succeeds. The saved model conversation then resumes from its previous point.

You should see status set to completed. The action_result contains one deterministic simulated execution receipt.

That is the critical boundary proven. Your authorized approval now completes the paused workflow without starting over.

Approval Did Not Behave as Expected?

• Check that events/approve-denied.json still uses the ops_requester role.
• Check that events/approve.json still uses the ops_approver role.
• Confirm that the workflow had awaiting_approval status before the allowed invocation.

Help me diagnose an unexpected OpsPilot approval response.

Replay requests and handle invalid input

Idempotency ties retries to stable identifiers. A repeated approval should return the saved result without executing the mutating tool again.

Before you replay the approval, do you expect a new receipt or the saved completed workflow? Hold that prediction.

• Replay the allowed approval event by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/approve.json --region us-east-1

How Is the Approval Replay Handled?

The workflow already has completed status. The approval handler returns its saved public view with duplicate set to true.

You should see the same execution receipt. No second disable_employee_access side effect is created.

Next, predict whether the original request ID creates another workflow or resolves to the existing one. Hold that prediction.

• Replay the original start event by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/start.json --region us-east-1

How Is the Request Replay Handled?

The stable request_id produces the same deterministic workflow ID. The matching payload hash proves that this is a retry of the same request.

You should see the existing completed workflow with duplicate set to true. The saved action result remains unchanged.

Handled validation gives malformed requests a predictable response. You will add one event whose action falls outside the supported event contract.

• Select the events folder in the Explorer sidebar of VS Code.
• Create invalid.json inside events.
• Paste this invalid-action event into events/invalid.json:

{"action":"launch_everything"}

What Does This Event Test?

The action value does not match ask, start, status, or approve. The Lambda should return a handled validation response.

• Save events/invalid.json.

Before you invoke this event, do you expect an unhandled failure or a structured validation response? Hold that prediction.

• Invoke the invalid event by running this command:

sam remote invoke AgentFunction --stack-name opspilot-agent-core --event-file events/invalid.json --region us-east-1

How Is Invalid Input Contained?

The Lambda routes the unsupported action through validation_error(). This keeps the failure inside the public response contract.

You should see status set to validation_error. The response explains that the action must use a supported value.

Replay Results Look Different?

• Check that every replay uses demo-offboarding-001 as the request_id.
• Compare events/start.json with the original request payload.
• Confirm that events/invalid.json contains only the unsupported action.

Help me troubleshoot an unexpected replay or validation result.

✔️ Awesome, I've got everything!

Your invalid event file is saved. Its payload now matches the handled validation test.

ⓧ I'd like to double check the full code

{"action":"launch_everything"}

What Should the File Contain?

The file contains one unsupported action value. This exact payload exercises the Lambda validation branch.

Correlate the JSON traces

Structured JSON events in Amazon CloudWatch Logs turn the workflow into an audit record. They expose model activity plus policy decisions without storing hidden chain-of-thought.

Before you inspect the logs, predict whether the denied approval and the successful mutation share the same workflow ID. Hold that prediction.

• Tail the OpsPilot log group by running this command:

aws logs tail /aws/lambda/opspilot-agent-core --since 10m --format json --region us-east-1

What Does This Log Check Cover?

• The log group targets the deployed OpsPilot Lambda.
• The time window covers the approval tests from this step.
• The output format preserves the structured fields used for correlation.

You should see an authorization_denied event for casey.requester. You should also see one mutating tool_executed event with its deterministic execution ID.

The model_turn entries correlate model_id, iteration, stop reason, and token counts. The Lambda log envelope adds its requestId.

The shared workflow_id connects policy decisions with approvals. The workflow_completed event proves the final outcome.

Missing Expected Trace Events?

• Repeat the denied approval invocation if its event falls outside the current log window.
• Repeat the allowed approval invocation to confirm that it returns the saved duplicate response.
• Check that the log group name is /aws/lambda/opspilot-agent-core.

Help me correlate missing OpsPilot trace events.

That is the core safety case proven. OpsPilot now has concrete evidence for every boundary you tested.

Secret mission

Add Incident-Aware Offboarding

Extend OpsPilot with a read-only incident-awareness tool. Amazon Nova Micro will decide when incident context matters while the existing approval boundary continues to protect access changes.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep the deployment available, pause active use, or delete the AWS resources. This stack can continue to generate variable usage and storage costs.

Cost warning

Calls to Amazon Bedrock can incur usage charges. Future invocations, stored workflow data, logs, and deployment artifacts can also contribute to your bill.

Monitoring and notifications from AWS Budgets are free. The Pause and Delete options below reduce your exposure to additional usage costs.

Resources you used:

• An AWS CloudFormation stack named opspilot-agent-core in us-east-1.
• An AWS Lambda function named opspilot-agent-core with reserved concurrency and JSON logging.
• An on-demand Amazon DynamoDB table containing persisted workflow messages, approvals, results, and traces.
• An Amazon CloudWatch Logs log group named /aws/lambda/opspilot-agent-core with three-day retention.
• An IAM execution role scoped to the workflow table and Amazon Nova Micro invocation.
• An account-wide AWS Budget named opspilot-agent-core-monthly with a $2 monthly limit and email notification.
• AWS SAM deployment artifacts associated with the stack.

Keep everything running

Keeping the deployment requires no cleanup. Choose this option when you want OpsPilot ready for more demonstrations or future development.

• Keep the opspilot-agent-core stack deployed in us-east-1.
• Avoid unnecessary remote invocations while the project is idle.
• Keep the account-wide $2 monthly budget alert active.
• Watch for the email notification when actual monthly cost is greater than 50% of $2.

Pause - I'll come back to this later

Pausing means ending active use while leaving the infrastructure ready for your return. Lambda has no idle compute charge, but persistent data, logs, artifacts, and future invocations can still contribute to costs.

• Stop sending remote invocations to the opspilot-agent-core stack.
• Leave the stack deployed so its workflow state and traces remain available.
• Keep the $2 account-wide budget alert active while the stack exists.

Delete - I don't want to use this again

Deleting the stack is decisive. Your local OpsPilot repository and commits stay untouched.

• Delete all stack-managed resources from the current PowerShell window by running this command:

sam delete --stack-name opspilot-agent-core --region us-east-1 --no-prompts

What does this command do?

The AWS SAM CLI targets the opspilot-agent-core stack in us-east-1. AWS CloudFormation removes the Lambda function, DynamoDB table, log group, IAM role, budget, and stack-managed deployment artifacts.

The --no-prompts option runs the deletion without interactive confirmation.

• Confirm the terminal reports that the stack deletion completed.

Stack deletion did not complete?

Keep your AWS session active. Read the terminal output for the resource that stopped cleanup.

Help me diagnose this stack deletion.

• End the short-lived AWS session by running this command:

aws logout

What does this command do?

The command removes cached login credentials for the default AWS CLI profile. Your AWS account and local OpsPilot repository remain.

• Confirm PowerShell returns to the prompt without reporting a failure.

That is a clean finish. Your deployed OpsPilot resources are removed while your local repository and commits remain ready for the next project.

Nice Work!

Nice Work!

Outstanding work! You've built the first deployable core of the long-lived OpsPilot platform.

You've learned how to:

• Use Amazon Nova Micro through Amazon Bedrock to select enterprise tools across multiple model turns.
• Enforce a deterministic approval boundary around mutating tools. Resume approved workflows from durable Amazon DynamoDB state.
• Prove replay-safe idempotency through repeated requests. Inspect structured JSON traces in Amazon CloudWatch Logs.
• Secret Mission: Add an incident-awareness tool that Amazon Nova Micro can select without changing the orchestration loop.

OpsPilot now demonstrates bounded agent autonomy through durable state, deterministic controls, and queryable evidence.

Ready to quiz yourself?