Add EventBridge scheduling: schedule_reminder, list_reminders, cancel_reminder

This commit is contained in:
daniel
2026-05-07 23:24:48 -05:00
parent 825294d433
commit 58ed60f7b7
46 changed files with 3605 additions and 24 deletions

View File

@@ -16,16 +16,16 @@
}
}
},
"c0db2a060be885d61722dfe6fbd3967e1c956826682078f6338123bf0c797e5b": {
"49f9e3ee598c0259165125872304200dbdffee263d76fca541a8630534d8f5c5": {
"displayName": "AgentRunner/Code",
"source": {
"path": "asset.c0db2a060be885d61722dfe6fbd3967e1c956826682078f6338123bf0c797e5b",
"path": "asset.49f9e3ee598c0259165125872304200dbdffee263d76fca541a8630534d8f5c5",
"packaging": "zip"
},
"destinations": {
"495395224548-us-east-1-82b9a17b": {
"495395224548-us-east-1-06bebbe8": {
"bucketName": "cdk-hnb659fds-assets-495395224548-us-east-1",
"objectKey": "c0db2a060be885d61722dfe6fbd3967e1c956826682078f6338123bf0c797e5b.zip",
"objectKey": "49f9e3ee598c0259165125872304200dbdffee263d76fca541a8630534d8f5c5.zip",
"region": "us-east-1",
"assumeRoleArn": "arn:${AWS::Partition}:iam::495395224548:role/cdk-hnb659fds-file-publishing-role-495395224548-us-east-1"
}
@@ -46,16 +46,31 @@
}
}
},
"a31aaa0bc9eab4fd6f17f10795fba05983dba0c88e83a263fe9fffe930da06b9": {
"8e7324457a5952eb51f04a34fbc5ba853252e7157d8d8958ac5fda92e72edb1f": {
"displayName": "Scheduler/Code",
"source": {
"path": "asset.8e7324457a5952eb51f04a34fbc5ba853252e7157d8d8958ac5fda92e72edb1f",
"packaging": "zip"
},
"destinations": {
"495395224548-us-east-1-89bca2fb": {
"bucketName": "cdk-hnb659fds-assets-495395224548-us-east-1",
"objectKey": "8e7324457a5952eb51f04a34fbc5ba853252e7157d8d8958ac5fda92e72edb1f.zip",
"region": "us-east-1",
"assumeRoleArn": "arn:${AWS::Partition}:iam::495395224548:role/cdk-hnb659fds-file-publishing-role-495395224548-us-east-1"
}
}
},
"c6cd323425a93776b45e2e0806064efbc5c84a3d6d78532282df6dd62cc14bda": {
"displayName": "AgentClawStack Template",
"source": {
"path": "AgentClawStack.template.json",
"packaging": "file"
},
"destinations": {
"495395224548-us-east-1-477d6bc7": {
"495395224548-us-east-1-51c91ff7": {
"bucketName": "cdk-hnb659fds-assets-495395224548-us-east-1",
"objectKey": "a31aaa0bc9eab4fd6f17f10795fba05983dba0c88e83a263fe9fffe930da06b9.json",
"objectKey": "c6cd323425a93776b45e2e0806064efbc5c84a3d6d78532282df6dd62cc14bda.json",
"region": "us-east-1",
"assumeRoleArn": "arn:${AWS::Partition}:iam::495395224548:role/cdk-hnb659fds-file-publishing-role-495395224548-us-east-1"
}

View File

@@ -32,7 +32,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:290:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:331:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -46,7 +46,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:294:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:335:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -60,7 +60,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:298:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:339:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -74,7 +74,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:303:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:344:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -88,7 +88,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:308:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:349:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -102,7 +102,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:313:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:354:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -116,7 +116,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:318:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:359:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -130,7 +130,7 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:323:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:364:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -144,7 +144,21 @@
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:328:5)",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:369:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
}
],
"/AgentClawStack/SchedulerLambdaArn": [
{
"type": "aws:cdk:logicalId",
"data": "SchedulerLambdaArn"
},
{
"type": "aws:cdk:creationStack",
"data": [
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:374:5)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
@@ -296,6 +310,36 @@
]
}
],
"/AgentClawStack/Scheduler/Resource": [
{
"type": "aws:cdk:logicalId",
"data": "SchedulerCFE73206"
},
{
"type": "aws:cdk:creationStack",
"data": [
"...new Function2 in aws-cdk-lib...",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:290:25)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
}
],
"/AgentClawStack/Scheduler/EventBridgeInvoke": [
{
"type": "aws:cdk:logicalId",
"data": "SchedulerEventBridgeInvoke72A0529A"
},
{
"type": "aws:cdk:creationStack",
"data": [
"...WrappedClass.addPermission in aws-cdk-lib...",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:303:17)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
}
],
"/AgentClawStack/CDKMetadata/Default": [
{
"type": "aws:cdk:logicalId",
@@ -516,6 +560,21 @@
]
}
],
"/AgentClawStack/Scheduler/ServiceRole/Resource": [
{
"type": "aws:cdk:logicalId",
"data": "SchedulerServiceRole62CDA70C"
},
{
"type": "aws:cdk:creationStack",
"data": [
"...new Function2 in aws-cdk-lib...",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:290:25)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
}
],
"/AgentClawStack/TgIngest/ServiceRole/DefaultPolicy/Resource": [
{
"type": "aws:cdk:logicalId",
@@ -611,5 +670,20 @@
"...node internals, ts-node, ts-node, ts-node..."
]
}
],
"/AgentClawStack/Scheduler/ServiceRole/DefaultPolicy/Resource": [
{
"type": "aws:cdk:logicalId",
"data": "SchedulerServiceRoleDefaultPolicyFA0D8235"
},
{
"type": "aws:cdk:creationStack",
"data": [
"...environmentFromArn.grantRead in aws-cdk-lib...",
"new AgentClawStack (/Users/daniel/agent-claw/cdk/lib/agent-claw-stack.ts:301:20)",
"<anonymous> (/Users/daniel/agent-claw/cdk/bin/agent-claw.ts:8:1)",
"...node internals, ts-node, ts-node, ts-node..."
]
}
]
}

View File

@@ -387,7 +387,7 @@
"Properties": {
"Code": {
"S3Bucket": "cdk-hnb659fds-assets-495395224548-us-east-1",
"S3Key": "c0db2a060be885d61722dfe6fbd3967e1c956826682078f6338123bf0c797e5b.zip"
"S3Key": "49f9e3ee598c0259165125872304200dbdffee263d76fca541a8630534d8f5c5.zip"
},
"Environment": {
"Variables": {
@@ -423,7 +423,7 @@
],
"Metadata": {
"aws:cdk:path": "AgentClawStack/AgentRunner/Resource",
"aws:asset:path": "asset.c0db2a060be885d61722dfe6fbd3967e1c956826682078f6338123bf0c797e5b",
"aws:asset:path": "asset.49f9e3ee598c0259165125872304200dbdffee263d76fca541a8630534d8f5c5",
"aws:asset:is-bundled": false,
"aws:asset:property": "Code"
}
@@ -872,6 +872,33 @@
"Effect": "Allow",
"Resource": "arn:aws:secretsmanager:us-east-1:495395224548:secret:agent-claw/google-credentials/*",
"Sid": "PerUserGoogleCredentialsReadRuntime"
},
{
"Action": [
"events:PutRule",
"events:PutTargets",
"events:ListRules",
"events:ListTargetsByRule",
"events:RemoveTargets",
"events:DeleteRule"
],
"Effect": "Allow",
"Resource": "arn:aws:events:us-east-1:*:rule/agent-claw-reminder-*",
"Sid": "EventBridgeScheduler"
},
{
"Action": [
"lambda:AddPermission",
"lambda:RemovePermission"
],
"Effect": "Allow",
"Resource": {
"Fn::GetAtt": [
"SchedulerCFE73206",
"Arn"
]
},
"Sid": "SchedulerLambdaPermission"
}
],
"Version": "2012-10-17"
@@ -1114,6 +1141,127 @@
"aws:asset:property": "Code"
}
},
"SchedulerServiceRole62CDA70C": {
"Type": "AWS::IAM::Role",
"Properties": {
"AssumeRolePolicyDocument": {
"Statement": [
{
"Action": "sts:AssumeRole",
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
}
}
],
"Version": "2012-10-17"
},
"ManagedPolicyArns": [
{
"Fn::Join": [
"",
[
"arn:",
{
"Ref": "AWS::Partition"
},
":iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
]
]
}
]
},
"Metadata": {
"aws:cdk:path": "AgentClawStack/Scheduler/ServiceRole/Resource"
}
},
"SchedulerServiceRoleDefaultPolicyFA0D8235": {
"Type": "AWS::IAM::Policy",
"Properties": {
"PolicyDocument": {
"Statement": [
{
"Action": [
"secretsmanager:GetSecretValue",
"secretsmanager:DescribeSecret"
],
"Effect": "Allow",
"Resource": "arn:aws:secretsmanager:us-east-1:495395224548:secret:agent-claw/telegram-bot-token-Oq3in3"
},
{
"Action": [
"events:RemoveTargets",
"events:DeleteRule"
],
"Effect": "Allow",
"Resource": "arn:aws:events:us-east-1:495395224548:rule/agent-claw-reminder-*"
}
],
"Version": "2012-10-17"
},
"PolicyName": "SchedulerServiceRoleDefaultPolicyFA0D8235",
"Roles": [
{
"Ref": "SchedulerServiceRole62CDA70C"
}
]
},
"Metadata": {
"aws:cdk:path": "AgentClawStack/Scheduler/ServiceRole/DefaultPolicy/Resource"
}
},
"SchedulerCFE73206": {
"Type": "AWS::Lambda::Function",
"Properties": {
"Code": {
"S3Bucket": "cdk-hnb659fds-assets-495395224548-us-east-1",
"S3Key": "8e7324457a5952eb51f04a34fbc5ba853252e7157d8d8958ac5fda92e72edb1f.zip"
},
"Environment": {
"Variables": {
"TELEGRAM_BOT_TOKEN_SECRET_ARN": "arn:aws:secretsmanager:us-east-1:495395224548:secret:agent-claw/telegram-bot-token-Oq3in3"
}
},
"FunctionName": "agent-claw-scheduler",
"Handler": "handler.handler",
"MemorySize": 128,
"Role": {
"Fn::GetAtt": [
"SchedulerServiceRole62CDA70C",
"Arn"
]
},
"Runtime": "python3.12",
"Timeout": 30
},
"DependsOn": [
"SchedulerServiceRoleDefaultPolicyFA0D8235",
"SchedulerServiceRole62CDA70C"
],
"Metadata": {
"aws:cdk:path": "AgentClawStack/Scheduler/Resource",
"aws:asset:path": "asset.8e7324457a5952eb51f04a34fbc5ba853252e7157d8d8958ac5fda92e72edb1f",
"aws:asset:is-bundled": false,
"aws:asset:property": "Code"
}
},
"SchedulerEventBridgeInvoke72A0529A": {
"Type": "AWS::Lambda::Permission",
"Properties": {
"Action": "lambda:InvokeFunction",
"FunctionName": {
"Fn::GetAtt": [
"SchedulerCFE73206",
"Arn"
]
},
"Principal": "events.amazonaws.com",
"SourceArn": "arn:aws:events:us-east-1:495395224548:rule/agent-claw-reminder-*"
},
"Metadata": {
"aws:cdk:path": "AgentClawStack/Scheduler/EventBridgeInvoke"
}
},
"CDKMetadata": {
"Type": "AWS::CDK::Metadata",
"Properties": {
@@ -1216,6 +1364,15 @@
"Arn"
]
}
},
"SchedulerLambdaArn": {
"Description": "Scheduler Lambda ARN — set as SCHEDULER_LAMBDA_ARN in agentcore.json",
"Value": {
"Fn::GetAtt": [
"SchedulerCFE73206",
"Arn"
]
}
}
},
"Parameters": {

View File

@@ -0,0 +1,268 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
# Per-invocation dedup: track sent message hashes to prevent AgentCore retry duplicates
_sent_hashes: set = set()
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
import hashlib
h = hashlib.md5(f'{chat_id}:{text}'.encode()).hexdigest()[:12]
if h in _sent_hashes:
print(f'[agent-runner] dedup: skipping duplicate message (hash={h})')
return
_sent_hashes.add(h)
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
try:
resp = urllib.request.urlopen(req, timeout=10)
resp_body = resp.read()
import re
msg_id = re.search(r'"message_id":(\d+)', resp_body.decode('utf-8', errors='replace'))
print(f'[agent-runner] Telegram sendMessage -> msg_id={msg_id.group(1) if msg_id else "?"} hash={h}')
except Exception as e:
print(f'[agent-runner] Telegram sendMessage FAILED: {type(e).__name__}: {e} hash={h}')
raise
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta ONLY
# Do NOT use event.get('data') — that's the full formatted summary,
# causing duplicate delivery alongside the token stream.
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
if not isinstance(delta, dict):
continue
token = delta.get('text', '')
if token:
text_buffer += token
# Only flush if buffer is very large — prevents splitting multi-turn responses
if len(text_buffer) > 1200:
print(f'[agent-runner] send chunk {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
print(f'[agent-runner] stream done buffer={len(text_buffer)} bot_token_set={bool(bot_token)}')
if text_buffer.strip() and bot_token:
print(f'[agent-runner] flushing {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,232 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
if body is not None:
for chunk in body.iter_chunks():
if not chunk:
continue
try:
event = json.loads(chunk.decode('utf-8'))
# Strands streaming event: 'data' field contains text delta
delta = event.get('data', '') or event.get('text', '')
if delta:
text_buffer += delta
# Flush on paragraph or sentence break, or if buffer is large
flush = (
text_buffer.rstrip().endswith(('\n\n', '.\n', '!\n', '?\n'))
or len(text_buffer) > 800
)
if flush and text_buffer.strip():
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
except (json.JSONDecodeError, UnicodeDecodeError):
pass
# Flush any remaining text
if text_buffer.strip() and bot_token:
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,263 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
# Per-invocation dedup: track sent message hashes to prevent AgentCore retry duplicates
_sent_hashes: set = set()
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
import hashlib, traceback as tb
h = hashlib.md5(f'{chat_id}:{text}'.encode()).hexdigest()[:12]
if h in _sent_hashes:
print(f'[agent-runner] dedup: skipping duplicate message (hash={h})')
print(f'[agent-runner] dedup stack: {tb.format_stack()[-3].strip()}')
return
_sent_hashes.add(h)
print(f'[agent-runner] SEND hash={h} text={repr(text[:40])}')
print(f'[agent-runner] SEND caller: {tb.format_stack()[-2].strip()}')
url = f'https://api.telegram.org/bot{token}/sendMessage'
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
if not isinstance(delta, dict):
continue
token = delta.get('text', '') or event.get('data', '')
if token:
text_buffer += token
flush = (
text_buffer.rstrip().endswith(('\n\n', '.\n', '!\n', '?\n'))
or len(text_buffer) > 800
)
if flush and text_buffer.strip():
print(f'[agent-runner] send chunk {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
print(f'[agent-runner] stream done buffer={len(text_buffer)} bot_token_set={bool(bot_token)}')
if text_buffer.strip() and bot_token:
print(f'[agent-runner] flushing {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,268 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
# Per-invocation dedup: track sent message hashes to prevent AgentCore retry duplicates
_sent_hashes: set = set()
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
import hashlib
h = hashlib.md5(f'{chat_id}:{text}'.encode()).hexdigest()[:12]
if h in _sent_hashes:
print(f'[agent-runner] dedup: skipping duplicate message (hash={h})')
return
_sent_hashes.add(h)
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
try:
resp = urllib.request.urlopen(req, timeout=10)
resp_body = resp.read()
import re
msg_id = re.search(r'"message_id":(\d+)', resp_body.decode('utf-8', errors='replace'))
print(f'[agent-runner] Telegram sendMessage -> msg_id={msg_id.group(1) if msg_id else "?"} hash={h}')
except Exception as e:
print(f'[agent-runner] Telegram sendMessage FAILED: {type(e).__name__}: {e} hash={h}')
raise
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('enrolled_services', user_profile.get('services', {})),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta ONLY
# Do NOT use event.get('data') — that's the full formatted summary,
# causing duplicate delivery alongside the token stream.
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
if not isinstance(delta, dict):
continue
token = delta.get('text', '')
if token:
text_buffer += token
# Only flush if buffer is very large — prevents splitting multi-turn responses
if len(text_buffer) > 1200:
print(f'[agent-runner] send chunk {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
print(f'[agent-runner] stream done buffer={len(text_buffer)} bot_token_set={bool(bot_token)}')
if text_buffer.strip() and bot_token:
print(f'[agent-runner] flushing {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,260 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
# Per-invocation dedup: track sent message hashes to prevent AgentCore retry duplicates
_sent_hashes: set = set()
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
import hashlib
h = hashlib.md5(f'{chat_id}:{text}'.encode()).hexdigest()[:12]
if h in _sent_hashes:
print(f'[agent-runner] dedup: skipping duplicate message (hash={h})')
return
_sent_hashes.add(h)
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta ONLY
# Do NOT use event.get('data') — that's the full formatted summary,
# causing duplicate delivery alongside the token stream.
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
if not isinstance(delta, dict):
continue
token = delta.get('text', '')
if token:
text_buffer += token
# Only flush if buffer is very large — prevents splitting multi-turn responses
if len(text_buffer) > 1200:
print(f'[agent-runner] send chunk {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
print(f'[agent-runner] stream done buffer={len(text_buffer)} bot_token_set={bool(bot_token)}')
if text_buffer.strip() and bot_token:
print(f'[agent-runner] flushing {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,244 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
# Extract text delta from contentBlockDelta
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
token = delta.get('text', '') or event.get('data', '')
if token:
text_buffer += token
flush = (
text_buffer.rstrip().endswith(('\n\n', '.\n', '!\n', '?\n'))
or len(text_buffer) > 800
)
if flush and text_buffer.strip():
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
if text_buffer.strip() and bot_token:
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,251 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
if not isinstance(delta, dict):
continue
token = delta.get('text', '') or event.get('data', '')
if token:
text_buffer += token
flush = (
text_buffer.rstrip().endswith(('\n\n', '.\n', '!\n', '?\n'))
or len(text_buffer) > 800
)
if flush and text_buffer.strip():
print(f'[agent-runner] send chunk {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
print(f'[agent-runner] stream done buffer={len(text_buffer)} bot_token_set={bool(bot_token)}')
if text_buffer.strip() and bot_token:
print(f'[agent-runner] flushing {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,246 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
token = delta.get('text', '') or event.get('data', '')
if token:
text_buffer += token
flush = (
text_buffer.rstrip().endswith(('\n\n', '.\n', '!\n', '?\n'))
or len(text_buffer) > 800
)
if flush and text_buffer.strip():
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
if text_buffer.strip() and bot_token:
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,29 @@
"""EventBridge-triggered Lambda: sends a Telegram reminder then deletes the rule."""
import json
import os
import boto3
import urllib.request
def handler(event, context):
chat_id = event['chat_id']
message = event['message']
rule_name = event['rule_name']
# Fetch bot token
sm = boto3.client('secretsmanager', region_name='us-east-1')
token = sm.get_secret_value(SecretId=os.environ['TELEGRAM_BOT_TOKEN_SECRET_ARN'])['SecretString']
# Send Telegram message
payload = json.dumps({'chat_id': chat_id, 'text': message}).encode()
req = urllib.request.Request(
f'https://api.telegram.org/bot{token}/sendMessage',
data=payload,
headers={'Content-Type': 'application/json'},
)
urllib.request.urlopen(req)
# Delete the one-time rule
eb = boto3.client('events', region_name='us-east-1')
eb.remove_targets(Rule=rule_name, Ids=['scheduler'])
eb.delete_rule(Name=rule_name)

View File

@@ -0,0 +1,196 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
_agentcore = boto3.client('bedrock-agentcore', region_name='us-east-1')
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Drain streaming response body (agent delivers to Telegram via send_message tool)
body = response.get('response')
if body is not None:
for _ in body.iter_chunks():
pass
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,261 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
# Per-invocation dedup: track sent message hashes to prevent AgentCore retry duplicates
_sent_hashes: set = set()
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
import hashlib
h = hashlib.md5(f'{chat_id}:{text}'.encode()).hexdigest()[:12]
if h in _sent_hashes:
print(f'[agent-runner] dedup: skipping duplicate message (hash={h})')
return
_sent_hashes.add(h)
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
if not isinstance(delta, dict):
continue
token = delta.get('text', '') or event.get('data', '')
if token:
text_buffer += token
flush = (
text_buffer.rstrip().endswith(('\n\n', '.\n', '!\n', '?\n'))
or len(text_buffer) > 800
)
if flush and text_buffer.strip():
print(f'[agent-runner] send chunk {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
print(f'[agent-runner] stream done buffer={len(text_buffer)} bot_token_set={bool(bot_token)}')
if text_buffer.strip() and bot_token:
print(f'[agent-runner] flushing {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -0,0 +1,263 @@
import json
import os
import time
import uuid
import boto3
import urllib.request
from typing import Any
# AWS clients
_ddb = None
_agentcore = None
def get_ddb():
global _ddb
if _ddb is None:
_ddb = boto3.resource('dynamodb')
return _ddb
def get_agentcore():
global _agentcore
if _agentcore is None:
from botocore.config import Config
_agentcore = boto3.client(
'bedrock-agentcore',
region_name='us-east-1',
config=Config(read_timeout=600, connect_timeout=10)
)
return _agentcore
def get_or_create_user(actor_id: str, from_info: dict) -> dict:
"""Look up user in registry, auto-registering on first contact."""
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return {'actor_id': actor_id, 'display_name': from_info.get('from_name', actor_id)}
table = get_ddb().Table(table_name)
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
if item:
return item
now = int(time.time())
item = {
'actor_id': actor_id,
'display_name': from_info.get('from_name') or actor_id,
'telegram_username': from_info.get('from_username', ''),
'created_at': str(now),
'status': 'pending',
'services': {},
}
table.put_item(Item=item)
print(f'[agent-runner] Registered new user (pending): {actor_id}')
return item
def update_user_status(actor_id: str, name: str, status: str) -> None:
table_name = os.environ.get('USERS_TABLE_NAME', '')
if not table_name:
return
table = get_ddb().Table(table_name)
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET display_name = :n, #s = :s',
ExpressionAttributeNames={'#s': 'status'},
ExpressionAttributeValues={':n': name, ':s': status},
)
# Per-invocation dedup: track sent message hashes to prevent AgentCore retry duplicates
_sent_hashes: set = set()
def send_telegram_direct(chat_id: str, token: str, text: str) -> None:
import hashlib
h = hashlib.md5(f'{chat_id}:{text}'.encode()).hexdigest()[:12]
if h in _sent_hashes:
print(f'[agent-runner] dedup: skipping duplicate message (hash={h})')
return
_sent_hashes.add(h)
url = f'https://api.telegram.org/bot{token}/sendMessage'
data = json.dumps({'chat_id': chat_id, 'text': text}).encode()
req = urllib.request.Request(url, data=data, headers={'Content-Type': 'application/json'})
urllib.request.urlopen(req, timeout=10)
def get_or_create_session(actor_id: str) -> str:
"""Look up active session for actor, or create a new one."""
table = get_ddb().Table(os.environ['SESSION_TABLE_NAME'])
response = table.get_item(Key={'actor_id': actor_id})
item = response.get('Item')
now = int(time.time())
ttl_8hr = now + (8 * 3600)
if item and item.get('ttl', 0) > now:
# Active session exists — extend TTL
table.update_item(
Key={'actor_id': actor_id},
UpdateExpression='SET #ttl = :ttl',
ExpressionAttributeNames={'#ttl': 'ttl'},
ExpressionAttributeValues={':ttl': ttl_8hr},
)
return item['session_id']
# Create new session
session_id = str(uuid.uuid4())
table.put_item(Item={
'actor_id': actor_id,
'session_id': session_id,
'created_at': str(now),
'ttl': ttl_8hr,
})
return session_id
def handler(event, context):
# ── Parse SQS records (FIFO — all from same actor) ───────────────────
records = []
for record in event.get('Records', []):
try:
records.append(json.loads(record['body']))
except (json.JSONDecodeError, KeyError):
continue
if not records:
return
first = records[0]
channel = first.get('channel', 'telegram')
chat_id = first.get('chat_id', '')
actor_id = f"{channel}:{chat_id}"
# ── User registry ─────────────────────────────────────────────────────
from_info = first.get('messages', [{}])[0]
user_profile = get_or_create_user(actor_id, from_info)
# ── Onboarding gate ─────────────────────────────────────────────────────
table_name = os.environ.get('USERS_TABLE_NAME', '')
if table_name and user_profile.get('status', 'active') == 'pending':
raw_prompt = records[0]['messages'][0]['text'] if records else ''
is_name_msg = bool(raw_prompt and len(raw_prompt.strip()) < 50 and '?' not in raw_prompt)
if is_name_msg:
name = raw_prompt.strip()
update_user_status(actor_id, name=name, status='active')
user_profile['display_name'] = name
user_profile['status'] = 'active'
prompt = f"[System: User just registered with name '{name}'. Welcome them warmly and ask how you can help.]"
else:
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
bot_token = ''
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
send_telegram_direct(chat_id, bot_token, "Hi! I don't recognize you yet. What's your name?")
return
# ── Get or create AgentCore session ──────────────────────────────────
session_id = get_or_create_session(actor_id)
print(f"[agent-runner] actor={actor_id} session={session_id} user={user_profile.get('display_name', '')}")
# ── Bundle messages ───────────────────────────────────────────────────
if len(records) == 1:
prompt = records[0]['messages'][0]['text']
else:
lines = [
f"[{i+1}] {r['messages'][0]['text']}"
for i, r in enumerate(records)
]
prompt = f"You have {len(records)} queued messages:\n" + "\n".join(lines)
# ── Build payload for AgentCore Runtime 1 ────────────────────────────
payload: dict[str, Any] = {
'prompt': prompt,
'actor_id': actor_id,
'session_id': session_id,
'user_profile': {
'display_name': user_profile.get('display_name', actor_id),
'telegram_username': user_profile.get('telegram_username', ''),
'google_email': user_profile.get('google_email', ''),
'allowed': user_profile.get('allowed', True),
'services': user_profile.get('services', {}),
},
'channel_adapter': {
'type': channel,
'target_id': str(chat_id),
'bot_token_secret_arn': os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', ''),
},
}
# ── Invoke AgentCore Runtime 1 ────────────────────────────────────────
runtime_arn = os.environ.get('RUNTIME_1_ARN', '')
if not runtime_arn or runtime_arn == 'PLACEHOLDER_SET_AFTER_RUNTIME_DEPLOY':
print(f"[agent-runner] RUNTIME_1_ARN not set — skipping AgentCore invoke")
print(f"[agent-runner] Would have sent: {json.dumps(payload)[:200]}")
return
client = get_agentcore()
response = client.invoke_agent_runtime(
agentRuntimeArn=runtime_arn,
runtimeSessionId=session_id,
payload=json.dumps(payload).encode(),
)
# Process streaming response: buffer text chunks and send to Telegram as paragraphs arrive
bot_token = ''
bot_token_secret_arn = os.environ.get('TELEGRAM_BOT_TOKEN_SECRET_ARN', '')
if bot_token_secret_arn:
sm = boto3.client('secretsmanager', region_name='us-east-1')
try:
bot_token = sm.get_secret_value(SecretId=bot_token_secret_arn)['SecretString']
except Exception as e:
print(f'[agent-runner] Failed to get bot token: {e}')
body = response.get('response')
text_buffer = ''
leftover = ''
if body is not None:
for raw_chunk in body.iter_chunks():
if not raw_chunk:
continue
# AgentCore streams SSE format: "data: {...}\n\n"
text = leftover + raw_chunk.decode('utf-8', errors='replace')
parts = text.split('\n\n')
leftover = parts[-1]
for part in parts[:-1]:
for line in part.splitlines():
if not line.startswith('data: '):
continue
data = line[6:].strip()
if not data or data == '[DONE]':
continue
try:
event = json.loads(data)
except (json.JSONDecodeError, ValueError):
continue
if not isinstance(event, dict):
continue
# Extract text delta from contentBlockDelta ONLY
# Do NOT use event.get('data') — that's the full formatted summary,
# causing duplicate delivery alongside the token stream.
delta = event.get('event', {}).get('contentBlockDelta', {}).get('delta', {})
if not isinstance(delta, dict):
continue
token = delta.get('text', '')
if token:
text_buffer += token
flush = (
text_buffer.rstrip().endswith(('\n\n', '.\n', '!\n', '?\n'))
or len(text_buffer) > 800
)
if flush and text_buffer.strip():
print(f'[agent-runner] send chunk {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
text_buffer = ''
# Flush any remaining text
print(f'[agent-runner] stream done buffer={len(text_buffer)} bot_token_set={bool(bot_token)}')
if text_buffer.strip() and bot_token:
print(f'[agent-runner] flushing {len(text_buffer)}c to {chat_id}')
send_telegram_direct(str(chat_id), bot_token, text_buffer.strip())
print(f"[agent-runner] Completed session={session_id} actor={actor_id}")

View File

@@ -18,7 +18,7 @@
"validateOnSynth": false,
"assumeRoleArn": "arn:${AWS::Partition}:iam::495395224548:role/cdk-hnb659fds-deploy-role-495395224548-us-east-1",
"cloudFormationExecutionRoleArn": "arn:${AWS::Partition}:iam::495395224548:role/cdk-hnb659fds-cfn-exec-role-495395224548-us-east-1",
"stackTemplateAssetObjectUrl": "s3://cdk-hnb659fds-assets-495395224548-us-east-1/a31aaa0bc9eab4fd6f17f10795fba05983dba0c88e83a263fe9fffe930da06b9.json",
"stackTemplateAssetObjectUrl": "s3://cdk-hnb659fds-assets-495395224548-us-east-1/c6cd323425a93776b45e2e0806064efbc5c84a3d6d78532282df6dd62cc14bda.json",
"requiresBootstrapStackVersion": 6,
"bootstrapStackVersionSsmParameter": "/cdk-bootstrap/hnb659fds/version",
"additionalDependencies": [

File diff suppressed because one or more lines are too long

View File

@@ -285,6 +285,47 @@ export class AgentClawStack extends cdk.Stack {
// NOTE: AgentCore runtime env vars are set in agentcore.json / agentcore deploy, not CDK.
// Output the URL so it can be manually set in agentcore.json OAUTH_START_URL.
// ── Lambda: scheduler ─────────────────────────────────────────────────
const schedulerFn = new lambda.Function(this, 'Scheduler', {
functionName: 'agent-claw-scheduler',
runtime: lambda.Runtime.PYTHON_3_12,
handler: 'handler.handler',
code: lambda.Code.fromAsset(path.join(__dirname, '../../src/lambdas/scheduler')),
timeout: cdk.Duration.seconds(30),
memorySize: 128,
environment: {
TELEGRAM_BOT_TOKEN_SECRET_ARN: telegramBotTokenSecretArn,
},
});
botTokenSecret.grantRead(schedulerFn);
// Allow EventBridge to invoke the scheduler Lambda
schedulerFn.addPermission('EventBridgeInvoke', {
principal: new iam.ServicePrincipal('events.amazonaws.com'),
sourceArn: `arn:aws:events:${this.region}:${this.account}:rule/agent-claw-reminder-*`,
});
// Allow scheduler Lambda to delete its own EventBridge rule
schedulerFn.addToRolePolicy(new iam.PolicyStatement({
actions: ['events:RemoveTargets', 'events:DeleteRule'],
resources: [`arn:aws:events:${this.region}:${this.account}:rule/agent-claw-reminder-*`],
}));
// AgentCore runtime: EventBridge + Lambda permissions for scheduling
runtime1Role.addToPolicy(new iam.PolicyStatement({
sid: 'EventBridgeScheduler',
actions: [
'events:PutRule', 'events:PutTargets',
'events:ListRules', 'events:ListTargetsByRule',
'events:RemoveTargets', 'events:DeleteRule',
],
resources: [`arn:aws:events:us-east-1:*:rule/agent-claw-reminder-*`],
}));
runtime1Role.addToPolicy(new iam.PolicyStatement({
sid: 'SchedulerLambdaPermission',
actions: ['lambda:AddPermission', 'lambda:RemovePermission'],
resources: [schedulerFn.functionArn],
}));
// ── Outputs ────────────────────────────────────────────────────────────
new cdk.CfnOutput(this, 'WorkspaceMcpFunctionUrl', {
@@ -329,5 +370,10 @@ export class AgentClawStack extends cdk.Stack {
value: runtime1Role.roleArn,
description: 'IAM execution role ARN for AgentCore Runtime 1',
});
new cdk.CfnOutput(this, 'SchedulerLambdaArn', {
value: schedulerFn.functionArn,
description: 'Scheduler Lambda ARN — set as SCHEDULER_LAMBDA_ARN in agentcore.json',
});
}
}