Building an Enterprise AI Operations Agent

A scheduler tells us where a workload failed. It usually cannot tell us why the business process failed.

Large enterprises rarely have one orchestration technology. Modern data pipelines may run in Airflow or Amazon MWAA, long-lived batch workloads may remain in Control-M, containerized processes may run on Amazon ECS, and older integration chains may still depend on shell scripts that invoke SnowSQL, APIs, SFTP utilities or internal applications.

When a workload fails, the first alert usually identifies the execution point: an Airflow task failed, a Control-M job ended with a non-zero return code, an ECS task stopped, or a shell wrapper returned an error. That is only the first layer of evidence.

A Control-M job may fail because a shell script failed. The shell script may have called SnowSQL. SnowSQL may expose a Snowflake query ID. Query history may finally reveal a schema mismatch. The scheduler is where the failure surfaced; it is not necessarily where the failure originated.

Evidence first. Reasoning second. Action last.

Six layers from failure signal to governed remediation

The architecture separates failure detection, source adapters, evidence collectors, normalized investigation state, reasoning and remediation. AgentCore is not required for this core flow.

Separate the reference implementation, the focused POC and the production target

A useful architecture article should make it obvious what has been designed, what exists as a lab/reference implementation, what still needs real-system validation, and what belongs to a later production evolution.

The agent should reduce investigation time, not duplicate the alert

Airflow, Control-M and monitoring systems already tell support teams that a workload failed. Repeating that notification in a chat interface adds little value. The agent becomes useful when it automatically performs the investigation steps a support engineer would otherwise perform manually across several consoles.

What a useful chat answer looks like

User:
Why did CUSTOMER_EDS_LOAD fail this morning?

Agent:
I traced the failed execution across three systems.

1. Control-M
   LOAD_CUSTOMER_EDS failed at 06:08.
   The failing command was load_customer_eds.sh; exit code 1.

2. Shell execution
   The script reached SnowSQL successfully.
   SnowSQL returned query ID 01c1....

3. Snowflake
   Query 01c1... failed with:
   invalid identifier 'CAMPAIGN_SOURCE'.

4. History
   I found a similar resolved incident associated with the
   same workload and schema-drift signature.

Assessment:
The scheduler and shell wrapper appear healthy.
The strongest evidence points to an upstream/schema-contract change.

Recommended next step:
Validate the upstream schema change and update the stage-to-EDS
mapping before retrying. An immediate retry is likely to fail again.

The answer is valuable only if the evidence was actually retrieved. The model should not manufacture the Control-M status, query ID, historical incident or remediation.

Previous incidents must come from an authoritative source — not from the LLM's memory

When a user asks “Has this happened before?” the agent needs an incident-search capability. Depending on the enterprise, that tool may query Jira, ServiceNow, another support platform or a curated operational-history table.

How incident matching can work

The current failure can produce a compact signature containing workload, failed component, error code, normalized error text, affected table/service and other correlation keys. The search tool can use exact filters first and semantic similarity second.

{
  "workload": "CUSTOMER_EDS_LOAD",
  "failure_class": "schema",
  "failed_component": "stage_to_eds",
  "platform": "snowflake",
  "object": "CUSTOMER_EDS",
  "error_signature": "invalid identifier CAMPAIGN_SOURCE"
}

The result returned to the agent should contain only fields it is authorized to see, such as incident ID, timestamps, root-cause classification, verified resolution, support owner and source reference.

Centralized operational history

At enterprise scale, I would consider persisting normalized operational metadata into a small observability schema rather than querying every historical system for every chat request.

OPS_OBSERVABILITY
├── WORKLOAD_RUN
├── WORKLOAD_STEP
├── FAILURE_EVIDENCE
├── INCIDENT_HISTORY
└── AGENT_INVESTIGATION

This creates a fast historical layer for questions such as “How many times has this failed in 90 days?”, “What were the most common root causes?”, or “Which resolution worked previously?” The agent can still drill into the source platform when raw evidence is required.

The first POC should prove the investigation path, not the entire enterprise platform

I would start with one narrow end-to-end scenario: MWAA → CloudWatch → Snowflake → LangGraph → Bedrock → read-only diagnosis. That is enough to demonstrate whether the agent genuinely reduces support effort.

What I would measure

Do not send every successful job — or every log line — to an LLM

The cheapest and most reliable architecture is event-driven and selective. Successful workloads should normally require no model call. Deterministic collectors should reduce the evidence before Bedrock is invoked.

Successful run
    → no AI investigation

Failure / SLA breach / repeated retry / DQ exception
    ↓
deterministic collectors
    ↓
filter relevant error window
    ↓
extract IDs + structured facts
    ↓
retrieve only necessary downstream evidence
    ↓
Bedrock reasoning
    ↓
evidence-backed diagnosis

A large CloudWatch log should not become a large prompt. A collector can select the failed task's time window, ERROR/exception records and correlation IDs, then pass a compact evidence package to the model. The same principle applies to Snowflake: query history deterministically and provide only the relevant rows.

Historical questions can use the normalized operational-history store first and drill into raw logs only when needed. This reduces repeated API calls, latency and token consumption.

Developer tools build the solution; production support uses the operations agent

Kiro, Claude Code and VS Code can help engineers build, test and evolve this application, but they are not the production-support interface. In production, the support analyst should interact with a dedicated operations-agent experience — initially a simple internal chat or test harness, and later an approved web, portal, Slack or Teams experience if the organization chooses.

Where MCP fits

MCP is not the chatbot and it is not tied to one IDE. It is a standardized tool-access protocol. During development, an MCP-capable environment such as Kiro or Claude Code can consume MCP tools. The production agent can also act as an MCP client independently of those developer tools.

FIRST POC

Support user / test harness
        ↓
Python + LangGraph
        ↓
Amazon Bedrock
        ↓
Direct tools
 ├─ MWAA API
 ├─ CloudWatch
 └─ Snowflake


ENTERPRISE EVOLUTION

Operations Chat UI
        ↓
Agent Application
        ↓
Amazon Bedrock
        ↓
Direct tools + MCP tools + RAG
        ↓
MWAA · CloudWatch · Snowflake · Jira · Runbooks · other governed systems

Detection and investigation are different concerns

The investigation can begin from an event-driven trigger—Airflow failure callback, ECS task-state event, Control-M alert, CloudWatch alarm—or from scheduled polling where event integration is not yet available.

{
  "event_type": "WORKLOAD_FAILURE",
  "source": "mwaa",
  "environment": "prod",
  "workload": "customer_daily_pipeline",
  "run_id": "scheduled__2026-08-29T06:00:00Z",
  "failed_unit": "stage_to_eds"
}

Airflow, Control-M, ECS and shell require different collectors

Airflow / Amazon MWAA

The MWAA adapter retrieves DAG-run and task-instance metadata, identifies the failed task, retry attempt, operator and timestamps, then locates task logs and downstream identifiers. Current MWAA supports programmatic Airflow REST API access, making this suitable for an external investigation service.

MWAA failure
  ↓ DAG run / task instance
failed task: stage_to_eds
  ↓ task log / CloudWatch
SnowflakeProgrammingError
query_id = 01bf...
  ↓ Snowflake evidence collector
actual SQL/data error

Control-M

For Control-M, the first useful artifacts are the job execution, run/order ID, command, host/agent, return code and job output/sysout. A job can then lead into a shell collector.

Control-M: LOAD_CUSTOMER_EDS FAILED
  ↓
/apps/customer/load_customer_eds.sh
exit code = 1
  ↓
Shell log: SnowSQL failed
query_id = 01c1...
  ↓
Snowflake Query History

Amazon ECS

The ECS adapter collects cluster, task ARN, task definition, container, stopped reason, exit code and CloudWatch log stream. The container log can expose a downstream API request ID, Snowflake query ID, S3 key or another correlation key.

Legacy shell

The shell adapter captures script identity, sanitized arguments, host, return code, stdout/stderr locations and child-system markers. Treating shell as a first-class execution source is important because many legacy schedulers ultimately launch scripts.

Follow evidence instead of querying every system

Every evidence item should carry provenance:

{
  "evidence_id": "ev-013",
  "type": "snowflake_query",
  "source_system": "snowflake",
  "retrieved_at": "2026-08-29T06:16:41Z",
  "correlation_key": "01c1...",
  "facts": {
    "status": "FAILED_WITH_ERROR",
    "error_code": "002003",
    "error_message": "invalid identifier 'CAMPAIGN_SOURCE'"
  }
}

Normalize heterogeneous systems before model reasoning

from typing import TypedDict, Literal

class Evidence(TypedDict):
    evidence_id: str
    type: str
    source_system: str
    correlation_key: str | None
    facts: dict

class FailureContext(TypedDict, total=False):
    request_id: str
    environment: str
    scheduler_type: Literal["airflow","controlm","ecs","shell","unknown"]
    workload_name: str
    run_id: str
    failed_unit: str
    evidence: list[Evidence]
    correlation_keys: dict
    next_lookup: str | None
    root_cause_hypothesis: dict | None
    confidence: float | None
    proposed_action: dict | None
    approval_status: str
    action_result: dict | None

This common contract means the reasoning layer does not need to understand every proprietary scheduler response format. It also makes recorded-evidence testing possible without connecting to production.

Use a stateful investigation graph, not an unconstrained autonomous loop

The investigation may discover new evidence, route to another collector, pause for approval and resume later. LangGraph is appropriate because the workflow is stateful and may need checkpoints and interrupts.

graph.add_node("route_source", route_source)
graph.add_node("collect_runtime", collect_runtime)
graph.add_node("extract_keys", extract_correlation_keys)
graph.add_node("collect_downstream", collect_downstream_evidence)
graph.add_node("reason", generate_root_cause_hypothesis)
graph.add_node("policy", evaluate_action_policy)
graph.add_node("approval", request_human_approval)
graph.add_node("execute", execute_governed_action)

The model correlates evidence; it does not own authorization

Bedrock can provide the model used for correlation, hypothesis generation and final synthesis. The model receives bounded evidence and approved operational knowledge.

The model should do

• correlate evidence across systems;
• separate confirmed facts from hypotheses;
• request missing evidence;
• rank likely root causes;
• produce an evidence-backed operational explanation.

The model should not do

• decide whether the caller is authorized to modify production;
• execute unrestricted SQL or shell commands;
• invent missing logs;
• bypass approvals;
• claim remediation succeeded without an action result.

SYSTEM:
You are an enterprise operations investigator.
Use only supplied evidence and approved knowledge.
For each root-cause statement, return supporting evidence IDs.
Label unsupported conclusions as hypotheses.
Ask for more evidence when confidence is insufficient.
Never claim an action completed unless a tool result proves it.

Use LangChain selectively

LangChain can reduce boilerplate for Bedrock/model adapters, tools, retrievers and structured output. It does not need to own the architecture.

class RootCauseAssessment(BaseModel):
    classification: Literal[
        "scheduler","runtime","data","schema",
        "permission","dependency","unknown"
    ]
    summary: str
    evidence_ids: list[str]
    confidence: float
    additional_evidence_required: list[str]
    recommended_action: str

Typed output prevents the next graph node from parsing a free-form paragraph.

Logs explain what happened; runbooks explain the approved response

Operational RAG can contain curated runbooks, known error patterns, recovery procedures, schema contracts, escalation rules and postmortem lessons. Retrieval should be driven by structured evidence such as platform, failed unit, error class and object name. Authorization filtering remains mandatory.

MWAA → CloudWatch → Snowflake

MWAA DAG: customer_daily_pipeline
  ↓ task: stage_to_eds
CloudWatch: SnowflakeProgrammingError
  query_id = 01bf...
  ↓
Snowflake Query History
  invalid identifier 'CAMPAIGN_SOURCE'
  ↓
Runbook: upstream schema drift procedure
  ↓
Root cause: schema contract mismatch
Confidence: 0.94

Control-M → shell → SnowSQL → Snowflake

Control-M
Job: LOAD_CUSTOMER_EDS
Status: FAILED
Command: /apps/customer/load_customer_eds.sh
Exit code: 1
      ↓
Shell log
SnowSQL failed
query_id=01c1...
      ↓
Snowflake Query History
Status: FAILED_WITH_ERROR
Error: invalid identifier 'CAMPAIGN_SOURCE'
      ↓
Root cause
EDS SQL is incompatible with the current upstream schema.

After evidence is normalized, the reasoning layer does not care whether the Snowflake query ID came from an Airflow task log or a legacy shell wrapper.

ECS → container → CloudWatch → downstream service

ECS task stopped
  ↓ DescribeTasks
container exitCode = 1
  ↓ CloudWatch
POST /campaign/enrich returned 503
request_id = api-87451
  ↓ dependency health tool
service unavailable
  ↓ runbook
retry after health recovery
  ↓ assessment
downstream dependency failure

Expose narrow business capabilities

tools/
├── airflow/
│   ├── get_dag_run.py
│   ├── get_task_instances.py
│   └── get_task_log_reference.py
├── controlm/
│   ├── get_job_run.py
│   └── get_job_output.py
├── ecs/
│   ├── get_task.py
│   └── get_container_log_reference.py
├── evidence/
│   ├── get_cloudwatch_logs.py
│   ├── get_snowflake_query.py
│   ├── get_load_history.py
│   └── inspect_s3_object.py
├── knowledge/
│   └── search_runbooks.py
└── action/
    ├── find_incident.py
    ├── create_incident.py
    └── request_retry.py

What is intentionally missing: generic shell execution, arbitrary SQL and unrestricted HTTP. Narrow tools create a smaller security and audit surface.

Correlation keys connect the enterprise execution chain

This is a major practical lesson: better upstream observability and consistent correlation IDs make the AI layer more deterministic and less expensive.

Autonomous investigation does not imply autonomous remediation

LangGraph interruption/resume fits well at the boundary between analysis and side effects.

The workflow must be safe to retry

An agent can fail after an external side effect succeeds. If an incident was created but the graph crashed before recording the result, retrying can create duplicates unless the action is idempotent.

dedup_key = sha256(
    f"{environment}:{workload}:{run_id}:{failure_signature}"
).hexdigest()

existing = find_incident(dedup_key)
if existing:
    return existing

return create_incident(summary, dedup_key)

MCP can standardize capabilities. AgentCore is optional.

The first implementation can run with LangGraph, Python, Bedrock, AWS APIs and Snowflake connectivity on normal application infrastructure. AgentCore becomes useful later when managed runtime, gateway, memory or observability capabilities provide operational value.

MCP

LangGraph Investigator
      ↓
MCP / governed capability layer
      ├── Airflow tools
      ├── Control-M tools
      ├── ECS tools
      ├── Snowflake tools
      ├── Runbook search
      └── Jira / incident tools

AgentCore later

• Runtime: managed hosting/execution option.
• Gateway: centralized access to tools/agents and MCP-oriented targets.
• Memory: managed memory where the use case needs it.
• Observability: CloudWatch-backed metrics/traces and OTEL-compatible telemetry.

The agent inherits enterprise boundaries; it does not bypass them

The investigator must itself be observable and measurable

Trace request/session ID, graph nodes, route decisions, tool authorization, tool latency, evidence IDs, model latency/tokens, retrieval document IDs, approval identity, idempotency keys and action outcomes. AgentCore Observability is one managed option; non-AgentCore OpenTelemetry instrumentation is also valid.

Before increasing autonomy, build a regression dataset of sanitized historical failures and measure root-cause classification accuracy, evidence precision, unsupported-claim rate, unnecessary-tool-call rate, time to diagnosis, token cost and human override rate.

Keep implementation understandable

enterprise-ai-operations-agent/
├── config/
│   ├── tool_policy.yaml
│   └── source_routes.yaml
├── src/
│   ├── agent/
│   │   ├── graph.py
│   │   ├── state.py
│   │   ├── prompts.py
│   │   └── reasoning.py
│   ├── adapters/
│   │   ├── mwaa.py
│   │   ├── controlm.py
│   │   ├── ecs.py
│   │   └── shell.py
│   ├── collectors/
│   │   ├── cloudwatch.py
│   │   ├── snowflake.py
│   │   ├── s3.py
│   │   └── application_api.py
│   ├── knowledge/
│   │   └── runbook_retriever.py
│   ├── policy/
│   │   ├── authorization.py
│   │   ├── risk.py
│   │   └── idempotency.py
│   └── actions/
│       ├── incident.py
│       └── retry.py
└── tests/
    ├── unit/
    ├── integration/
    └── evaluation/

Build depth before breadth

What I deliberately would not do

The valuable agent is not the one that talks the most. It is the one that can show how it knows.

An enterprise operations agent becomes credible when it can trace a failure through heterogeneous runtime systems, show the evidence behind its conclusion, distinguish facts from hypotheses, preserve state, honor authorization and keep remediation behind governed controls.

The first practical implementation can be small: MWAA → CloudWatch → Snowflake → LangGraph → Bedrock → read-only diagnosis. Once that path is proven, the same normalized failure model can expand to Control-M, legacy shell wrappers, ECS, governed actions, MCP and optional AgentCore services.

Official documentation

1. Amazon MWAA — Apache Airflow REST API
2. Amazon MWAA — service and CloudWatch integration
3. Amazon Bedrock AgentCore — Observability
4. AgentCore Observability — getting started
5. AgentCore Gateway / Runtime targets
6. LangGraph — Persistence
7. LangGraph — Interrupts