Automating Operational Workflows with AWS DevOps Agent Custom Agents, Amazon EventBridge, and AWS Lambda

```python
import boto3, json, os, time, base64, zipfile, io, urllib.request
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest

REGION = os.environ.get('AWS_REGION', 'us-east-1')
SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN'] # e.g. arn:aws:sns:us-east-1:123456789012:my-topic
AGENT_NAME = os.environ.get('AGENT_NAME', 'my-custom-agent')
ARTIFACT_ID = os.environ.get('ARTIFACT_ASSET_ID', '') # stable ki-* ID, set once after first run

POLL_INTERVAL = 30 # seconds between status checks
MAX_POLLS = 25 # 25 × 30s = ~12.5 min maximum wait

devops = boto3.client('devops-agent', region_name=REGION)
sns = boto3.client('sns', region_name=REGION)

def lambda_handler(event, context):
# 1. Extract taskId from the CloudTrail event
task = event['detail']['responseElements']['task']
task_id = task['taskId']
agent_space_id = task['agentSpaceId']
created_at = task.get('createdAt', 'N/A')

# 2. Poll GetBacklogTask until the agent reaches a terminal state
terminal = {'COMPLETED', 'FAILED', 'TIMED_OUT', 'CANCELLED'}
detail = {}
for i in range(MAX_POLLS):
resp = devops.get_backlog_task(agentSpaceId=agent_space_id, taskId=task_id)
detail = resp.get('backlogTask') or resp.get('task') or {}
status = detail.get('status', 'UNKNOWN')
print(f"Poll {i+1}: {status}")
if status in terminal:
break
time.sleep(POLL_INTERVAL)

execution_id = detail.get('executionId')
updated_at = detail.get('updatedAt', 'N/A')

# 3. Fetch the artifact via Data Plane API (preferred)
report = '(No output)'
if status == 'COMPLETED' and ARTIFACT_ID:
try:
artifact = get_asset_content(agent_space_id, ARTIFACT_ID)
report = format_artifact(artifact)
except Exception as e:
print(f"Asset API failed: {e}, falling back to journal")
if execution_id:
report = get_summary_from_journal(agent_space_id, execution_id)
elif status == 'COMPLETED' and execution_id:
report = get_summary_from_journal(agent_space_id, execution_id)

# 4. Take action — email via SNS (swap for Jira, Slack, Step Functions, etc.)
sns.publish(
TopicArn=SNS_TOPIC_ARN,
Subject=f"[{AGENT_NAME}] {status}"[:100],
Message=(
f"Agent: {AGENT_NAME}\n"
f"Status: {status}\n"
f"Started: {created_at}\n"
f"Completed: {updated_at}\n\n"
f"{report}"
)
)
return {'statusCode': 200, 'body': 'OK'}

def get_asset_content(agent_space_id, asset_id):
"""Fetch artifact via DevOps Agent Data Plane API (dp. endpoint)."""
url = f'https://dp.aidevops.{REGION}.api.aws/asset/agent-space/{agent_space_id}/assets/{asset_id}/content'
credentials = boto3.session.Session().get_credentials().get_frozen_credentials()
request = AWSRequest(method='GET', url=url)
SigV4Auth(credentials, 'aidevops', REGION).add_auth(request)
resp = urllib.request.urlopen(urllib.request.Request(url, headers=dict(request.headers)), timeout=15)
body = json.loads(resp.read())
zip_bytes = base64.b64decode(body['content']['zipFile'])
z = zipfile.ZipFile(io.BytesIO(zip_bytes))
for name in z.namelist():
return json.loads(z.read(name).decode('utf-8'))
return None

def format_artifact(artifact):
"""Convert artifact JSON elements into readable text for email."""
lines = []
for el in artifact.get('elements', []):
if el.get('type') == 'text':
lines.append(el.get('content', ''))
elif el.get('type') in ('data_table', 'table'):
lines.append(f"\n### {el.get('title', 'Table')}")
cols = el.get('columns', [])
headers = [c.get('label', c.get('key', '')) for c in cols]
lines.append(' | '.join(headers))
lines.append(' | '.join(['---'] * len(headers)))
for row in el.get('data', [])[:20]:
lines.append(' | '.join(str(row.get(c.get('key', ''), '')) for c in cols))
elif el.get('type') == 'chart':
lines.append(f"\n### {el.get('title', 'Chart')}")
for d in el.get('data', []):
vals = list(d.values())
lines.append(f" {vals[0]}: {vals[1] if len(vals) > 1 else ''}")
lines.append('')
return '\n'.join(lines)

def get_summary_from_journal(agent_space_id, execution_id):
"""Fallback: extract final assistant text from journal records."""
records, kwargs = [], {'agentSpaceId': agent_space_id, 'executionId': execution_id}
while True:
r = devops.list_journal_records(**kwargs)
batch = r.get('journalRecords') or r.get('records') or []
records.extend(batch)
if not r.get('nextToken'):
break
kwargs['nextToken'] = r['nextToken']
texts = []
for r in sorted(records, key=lambda x: x.get('createdAt', '')):
if r.get('recordType') != 'message':
continue
try:
msg = json.loads(r.get('content', '{}'))
if msg.get('role') == 'assistant':
for item in msg.get('content', []):
if item.get('type') == 'text' and item.get('text', '').strip():
texts.append(item['text'].strip())
except Exception:
pass
return texts[-1] if texts else '(No output found)'
```