7
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@PDC-Kurashinak(くらっち)
in

【AWS】LINEからStrands Agentsを呼び出す！

7
Last updated at Posted at 2026-05-14

どうもこんにちは。
今回は、スマホのLINEアプリからAmazon Bedrock AgentCore Runtime上のStrands Agentsを呼び出す構成を作ったので、記事にします。

経緯

ここ半年でOpenCrawというオープンソースの自律型AIパーソナルエージェントが話題になりました。OpenCrawを常時動かすために、MacMiniを購入される方がいたり、EC2インスタンスへデプロイしたりというような使い方をされる方が多くいらっしゃった認識です。

私もOpenCrawを動かしたかったですが、金銭的に厳しいなというのがあったので、サーバレスで動かしたいなと考え、Bedrock AgentCore Runtimeで代替してAIパーソナルエージェントを構築しました。（まだ不完全だけど...）

構成

以下のように、LINEを起点として、API Gateway, Lambda, Bedrock AgentCore Runtime, DynamoDBを用いた最小構成としています。

スクリーンショット 2026-04-30 23.50.19.png

  • DynamoDBは、後々会話履歴以外のデータを保管するために用意しています。（今回は保存したデータを参照するようにはしていません。）
  • LINE APIに必要なアクセストークンなどは、SecretsManagerに保管するようにしています。

リポジトリ

こちらのリポジトリをgit cloneしている想定で記事を書いています。

流れ

以下の流れで構築を行なっていきます。

  1. LINE Official Account を作成する
  2. Messaging API を有効化する
  3. JWT を使って channel access token v2.1 を発行する
  4. CDK で API Gateway / Lambda / DynamoDB / AgentCore Runtime をデプロイする
  5. LINE Webhook から AgentCore Runtime を同期呼び出しして返信する

前提条件

このリポジトリの検証に必要な環境は以下です。

  • Python 3.12
    • 仮想環境を使用してpip installなどのセットアップを行います
    • strands-agents と bedrock-agentcore は Python 3.10 以上が必要です
  • Docker
  • AWS CLI
  • AWS CDK CLI
  • LINE Developers コンソールへのアクセス権限

また、AWSのリージョンはap-northeast-1、BedrockモデルIDはjp.anthropic.claude-sonnet-4-6を指定します。

構築手順

0. Git Clone

以下を実行しておいてください。

git clone https://github.com/kuracchi-enj/bedrock-agentcore--runtime-from-line.git

cd bedrock-agentcore--runtime-from-line

1. LINE Official Account と Messaging API の準備

以下のページから、LINE公式アカウントのセットアップを行います。

ご自身のLINEアカウントでセットアップできます。

また、セットアップができたら、設定からMessaging APIを有効にします。Messaging APIを有効にすると、Messaging API用のチャンネルが作成されます。作成されたチャンネルの以下の情報を控えておいてください。

  • チャンネルID
  • チャンネルシークレット

作成されたことを確認したら、以下へアクセスして、LINE Developerコンソールへ遷移します。

2. JWTを生成して、チャンネルアクセストークンを発行する

LINEのチャンネルアクセストークンを発行するために、JWTを生成します。
まずは以下のコマンドを実行して、秘密鍵と公開鍵を出力します。

cd /path/to/bedrock-agentcore--runtime-from-line

python3.12 -m venv .venv
source .venv/bin/activate

python -m pip install -r requirements.txt
python generate_line_signing_key.py

出力された鍵情報は漏洩しないようにしてください。

次に、LINE Developer コンソールへ移動し、チャンネルの基本設定画面を開きます。そこに アサーション署名キー という項目があるので、そこの「公開鍵を登録する」ボタンをクリックします。

image.png

そこで、先ほど出力した公開鍵をコピペして登録します。登録すると、キーが返ってくるので、コピーして控えておきます。（このキーはkidと呼びます。）

次にJWTの生成を行います。以下のコマンドを実行してください。

cp .env.example .env
vi .env

vimを使って環境変数を用意します。

# チャンネルID
LINE_CHANNEL_ID=1234567890
# kid
LINE_ASSERTION_SIGNING_KEY_ID=your-kid
# 秘密鍵
LINE_PRIVATE_KEY_JWK={"kty":"RSA","alg":"RS256","use":"sig","e":"AQAB","n":"...","d":"...","p":"...","q":"...","dp":"...","dq":"...","qi":"..."}

# -----------------JWTの生成には使用しない---------------------
# チャンネルシークレット
LINE_CHANNEL_SECRET=your-line-channel-secret
# チャンネルアクセストークン
LINE_CHANNEL_ACCESS_TOKEN=your-line-channel-access-token

環境変数を用意したら、以下のコマンドを実行します。

source .venv/bin/activate
python generate_jwt.py

これによって、以下のJWTが作られます。

header:

項目 値
alg RS256
typ JWT
kid {{assertion signing key id}}

payload:

項目 値
iss {{channel id}}
sub {{channel id}}
aud https://api.line.me/
exp 現在時刻 + 30分以内
token_exp 30日以内

上記の情報が揃ったら、チャンネルアクセストークンを発行します。

JWT_ASSERTION="<generate_jwt.py の出力>"

curl -X POST https://api.line.me/oauth2/v2.1/token \
  -H "Content-Type: application/x-www-form-urlencoded" \
  --data-urlencode "grant_type=client_credentials" \
  --data-urlencode "client_assertion_type=urn:ietf:params:oauth:client-assertion-type:jwt-bearer" \
  --data-urlencode "client_assertion=${JWT_ASSERTION}"

上記を実行したら、以下のようなレスポンスが返ってきます。

{
  "access_token": "....",
  "expires_in": 2592000,
  "token_type": "Bearer",
  "key_id": "...."
}

3. 機密情報をSecretsManagerに格納する

access_tokenの項目に、チャンネルアクセストークンが出力されるので、SecretsManagerに格納します。格納はAWS CLIで行なっても、AWS マネジメントコンソール上で行なってもどちらでも構いません。

チャンネルアクセストークン格納

aws secretsmanager create-secret \
  --name line-webhook-receiver/channel-access-token \
  --secret-string '<LINE_CHANNEL_ACCESS_TOKEN>' \
  --region ap-northeast-1

更新する場合は以下を実行します。

aws secretsmanager put-secret-value \
  --secret-id line-webhook-receiver/channel-access-token \
  --secret-string '<LINE_CHANNEL_ACCESS_TOKEN>' \
  --region ap-northeast-1

チャンネルシークレット格納

チャンネルシークレットもSecretsManagerに格納します。

aws secretsmanager create-secret \
  --name line-webhook-receiver/channel-secret \
  --secret-string '<LINE_CHANNEL_SECRET>' \
  --region ap-northeast-1

更新する場合は以下を実行します。

aws secretsmanager put-secret-value \
  --secret-id line-webhook-receiver/channel-secret \
  --secret-string '<LINE_CHANNEL_SECRET>' \
  --region ap-northeast-1

これでLINE APIを使用するための認証の準備が整いました。

4. Strands Agents × CDK の環境を用意する

ここからはアプリケーションコードをローカルで動かしつつ、最終的にAWSへデプロイしていきます。

まずは依存関係を入れます。

python -m pip install -r strands_agent/requirements.txt
python -m pip install -r infra/requirements.txt

strands-agents と bedrock-agentcore は Python 3.10 以上が必要です。私の手元では最初 Python 3.9 で仮想環境を作ってしまい、依存解決に失敗しました。素直に Python 3.12 を使うのがよいです。

次に、AWSの利用リージョンとBedrockモデルIDを環境変数に入れます。

export AWS_REGION=ap-northeast-1
export AWS_DEFAULT_REGION=ap-northeast-1
export BEDROCK_MODEL_ID=jp.anthropic.claude-sonnet-4-6

また、AWS CLI が利用できることも確認します。

aws sts get-caller-identity

5. Strands Agent をローカルで動作確認する

デプロイ前に、Strands Agent 単体がちゃんと応答することを確認しておきます。

このリポジトリでは、strands_agent/local_chat.py から標準入力の1メッセージを Strands Agent に渡し、標準出力に返すようにしています。

printf '今日やることを3つに整理して' | python -m strands_agent.local_chat

例えば、以下のように雑に投げても日本語で返ってきます。

printf '冷蔵庫に卵と豆腐があります。夕食案を考えて' | python -m strands_agent.local_chat

今回の構成では jp.anthropic.claude-sonnet-4-6 を使っています。

6. Docker で AgentCore Runtime 用コンテナを確認する

Bedrock AgentCore Runtime 上でも同じアプリケーションを動かしたいので、Strands Agent を Docker 化しています。

ビルドは以下です。

docker build --platform linux/arm64 -t life-os-agent:latest -f strands_agent/Dockerfile strands_agent

起動は以下です。

docker run --rm -p 8080:8080 \
  -e AWS_REGION=ap-northeast-1 \
  -e AWS_DEFAULT_REGION=ap-northeast-1 \
  -e BEDROCK_MODEL_ID=jp.anthropic.claude-sonnet-4-6 \
  -e AWS_PROFILE=default \
  -v ~/.aws:/root/.aws:ro \
  life-os-agent:latest

別ターミナルから invocation します。

curl -i -X POST http://localhost:8080/invocations \
  -H 'Content-Type: application/json' \
  -d '{"prompt":"今日やることを3つに整理して"}'

レスポンスが 200 OK で返ってくれば、AgentCore Runtime コンテナとして最低限の導線はOKです。

7. Lambda ロジックをローカルテストする

infra/lambda_processor/handler.py には、以下の責務があります。

  • incoming event を DynamoDB に保存する
  • テキストメッセージだけを AgentCore Runtime に渡す
  • 応答を LINE Reply API に返す
  • outgoing event を DynamoDB に保存する
  • AgentCore が失敗したら固定のフォールバック文言を返す

このロジックはユニットテストを用意しています。

python -m unittest tests/test_lambda_processor.py

このテストでは、以下を検証しています。

  • 正常時は AgentCore の返答がそのまま LINE 返信に使われる
  • AgentCore 失敗時はフォールバック文言へ切り替わる
  • 非テキストイベントは保存だけ行い、返信しない
  • AGENTCORE_SESSION_NAMESPACE を使ったセッションIDの合成

8. CDK で AWS インフラをデプロイする

このリポジトリでは CDK を使って、以下のリソースを一括で作成します。

  • API Gateway
  • Lambda1 line-webhook-receiver
  • Lambda2 line-webhook-processor
  • DynamoDB line-conversation-events
  • Bedrock AgentCore Runtime
  • AgentCore Runtime 実行用 IAM Role

8-1. bootstrap

BootStrapを行なっていない場合は、先に以下を実行します。

cdk bootstrap

8-2. synth

CloudFormationスタックが生成されることを確認します。

cdk synth

cdk.json では python infra/app.py を実行するようにしています。ルートの .venv を有効化した状態で実行してください。

8-3. deploy

cdk deploy

今回の構成では、AgentCore Runtime 用コンテナを DockerImageAsset で publish しています。そのため、docker tag や docker push を手で打たなくても、cdk deploy の中で以下が実行されます。

  • strands_agent/ の Docker build
  • CDK asset 用 ECR への push
  • AgentCore Runtime への image URI 設定

デプロイが完了すると、CloudFormation の output として以下が出ます。

  • LineWebhookUrl
  • LifeOsAgentCoreRuntimeArn
  • LifeOsAgentCoreRuntimeId
  • LifeOsAgentCoreRuntimeVersion

9. LINE Webhook URL を設定する

デプロイが完了したら、LINE Developers コンソールの Messaging API タブを開き、Webhook URL に CloudFormation output の LineWebhookUrl を設定します。

image.png

例えば以下のようなURLになります。

https://xxxxxxxxxx.execute-api.ap-northeast-1.amazonaws.com/prod/line/webhook

設定したら、Webhookの利用 を有効にします。

その後、コンソール上の 検証 ボタンを使って webhook が疎通することを確認します。

10. デプロイ後の確認

AgentCore Runtime を直接叩いて確認する

まずは LINE を経由せず、AgentCore Runtime を直接叩くと切り分けがしやすいです。

aws bedrock-agentcore invoke-agent-runtime \
  --agent-runtime-arn <LifeOsAgentCoreRuntimeArn> \
  --runtime-session-id test-session-v2-20260430-000000001 \
  --content-type application/json \
  --accept application/json \
  --payload '{"prompt":"こんにちは"}' \
  /tmp/agentcore-output.json \
  --region ap-northeast-1

返却内容の確認:

cat /tmp/agentcore-output.json

私の環境では以下のようなJSONが返ってきました。

{"appName": "life-os-japanese-assistant", "message": "こんにちは！😊\n今日はどんなことでお手伝いできますか？"}

LINE から実際に送る

その後、スマホの LINE アプリから bot にメッセージを送り、以下を確認します。

  • LINE に応答が返ってくる

image.png

  • DynamoDB に incoming が保存される
  • DynamoDB に outgoing が保存される

確認例:

aws dynamodb scan \
  --table-name line-conversation-events \
  --max-items 10 \
  --region ap-northeast-1

AWS インフラ構成の詳細解説

ここからは、なぜこの構成にしているのかをもう少し詳しく書きます。

API Gateway

POST /line/webhook だけを受ける、シンプルなREST APIです。

LINE の Webhook は HTTPS の公開エンドポイントが必要なので、ここでは素直に API Gateway を使っています。

Lambda1: line-webhook-receiver

この Lambda は Webhook の受け口です。主な責務は以下です。

  • x-line-signature を取り出す
  • request body を チャンネルシークレット で HMAC-SHA256 検証する
  • 検証OKなら Lambda2 に同期で委譲する

ポイントは、「署名検証の前に body をいじらない」ことです。JSON parse した後で再シリアライズすると、署名検証に失敗します。

コードとしては、以下のように x-line-signature と request body を使って HMAC を計算しています。

def is_valid_signature(body: str, signature: str) -> bool:
    channel_secret = get_secret("LINE_CHANNEL_SECRET_NAME")
    digest = hmac.new(
        channel_secret.encode("utf-8"),
        body.encode("utf-8"),
        hashlib.sha256,
    ).digest()
    expected_signature = base64.b64encode(digest).decode("utf-8")
    return hmac.compare_digest(expected_signature, signature)

ここでやっていることはシンプルで、Secrets Manager から channel secret を取り出し、その値を使って body を署名検証しているだけです。

署名が正しい場合だけ、以下のように Lambda2 へ同期で委譲します。

response = lambda_client.invoke(
    FunctionName=os.environ["PROCESSOR_FUNCTION_NAME"],
    InvocationType="RequestResponse",
    Payload=json.dumps(payload).encode("utf-8"),
)

この構成にしている理由は、署名検証と会話ロジックを分けたいからです。LINE 由来の検証ロジックは Lambda1 に閉じ込めて、Lambda2 は「正しいイベントを受け取った後の処理」に集中させています。

Lambda2: line-webhook-processor

ここが実質的な業務ロジックです。

やっていることは以下です。

  1. LINE の incoming event を DynamoDB に保存
  2. テキストメッセージだけを AgentCore Runtime に同期送信
  3. 応答文を LINE Reply API で返す
  4. outgoing event を DynamoDB に保存

処理の流れをコードで見ると、まず incoming event を必ず保存しています。

for line_event in events:
    handled += 1
    store_incoming_event(request_id, destination, line_event)

その後、テキストメッセージだけを対象にしています。

if line_event.get("type") != "message":
    continue

message = line_event.get("message") or {}
if message.get("type") != "text":
    continue

この段階で、画像やスタンプ、follow イベントのような非テキストイベントは保存だけ行ってスキップされます。

AgentCore Runtime の呼び出しは以下です。

response = agentcore_client().invoke_agent_runtime(
    agentRuntimeArn=runtime_arn,
    runtimeSessionId=runtime_session_id,
    contentType="application/json",
    accept="application/json",
    payload=json.dumps({"prompt": text}).encode("utf-8"),
)

payload は今のところかなり最小で、prompt に LINE のメッセージ本文だけを渡しています。

返ってきたレスポンスからは、以下のように message を取り出します。

def parse_agentcore_response_body(body: str) -> str:
    payload = json.loads(body)
    if isinstance(payload, dict):
        if isinstance(payload.get("message"), str) and payload["message"].strip():
            return payload["message"].strip()

つまり AgentCore Runtime コンテナ側は、{"message": "<応答本文>"} の形で返せばよいようにしてあります。

返信は LINE Reply API に対して直接 POST しています。

req = request.Request(
    "https://api.line.me/v2/bot/message/reply",
    data=payload,
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {channel_access_token}",
    },
    method="POST",
)

ここでアクセストークンは環境変数ではなく Secrets Manager から取り出しています。

また、AgentCore の呼び出しに失敗した場合でも LINE 側には何かしら返したいので、フォールバック文言を用意しています。

FALLBACK_REPLY_TEXT = "今は応答を生成できませんでした。少し時間をおいてもう一度送ってください。"

この設計にしておくと、Runtime 側の一時障害でユーザーに完全無応答を返すのを避けられます。

DynamoDB

DynamoDB は現時点では「履歴保存と監査」のためにだけ使っています。

テーブルは以下のようなキー設計です。

  • partition key: pk
  • sort key: sk

例えば USER#<line_user_id> を partition key にして、時系列で incoming / outgoing を並べています。

将来的には以下のような用途に広げやすいです。

  • 会話履歴の読み出し
  • 家計簿データの保存
  • タスクや予定データの保存

Bedrock AgentCore Runtime

Agent 本体は Lambda ではなく AgentCore Runtime に載せています。

理由は以下です。

  • Strands Agent をそのまま実行しやすい
  • コンテナとしてローカルと本番を揃えやすい
  • 将来的に tool や複雑な会話ロジックを追加しやすい

今回は BedrockAgentCoreApp ベースで、最小限の POST /invocations を受けるアプリにしています。

Strands Agent

Agent の system prompt は「生活 OS 向けの汎用日本語会話アシスタント」です。

この時点では、以下はまだ入れていません。

  • 外部 tool
  • AgentSkills
  • AgentCore Memory
  • 家計簿管理/スケジュール管理の専用ロジック

まずは「LINE から自然な日本語会話が返ってくる」ことを確認するのが目的だからです。

コードとしては、strands_agent/agent.py で Agent を組み立てています。

def create_agent() -> Agent:
    session = boto3.session.Session(region_name=get_region())
    model = BedrockModel(
        model_id=get_model_id(),
        boto_session=session,
    )
    return Agent(
        model=model,
        system_prompt=get_system_prompt(),
        callback_handler=None,
    )

ポイントは以下です。

  • モデルは BedrockModel を使う
  • model ID と region は環境変数から切り替え可能
  • callback_handler=None にして、Strands のデフォルトの標準出力ストリームを止める

特に callback_handler=None は地味ですが重要で、これを入れないとローカル実行時に標準出力へストリーミングされて、後段で整形して print したテキストと二重に見えることがありました。

レスポンスの整形は extract_text() で行っています。

def extract_text(response) -> str:
    message = getattr(response, "message", None)
    if not isinstance(message, dict):
        return str(response)

    content = message.get("content") or []
    texts: list[str] = []
    seen: set[str] = set()

ここでは message.content の中から text だけを拾い、重複を除外しています。Strands のレスポンス構造をそのまま LINE に返すのではなく、最終的なテキストだけを抜き出すためです。

AgentCore Runtime entrypoint

AgentCore Runtime から呼ばれる entrypoint は strands_agent/runtime_app.py にあります。

app = BedrockAgentCoreApp()
agent = create_agent()


@app.entrypoint
def invoke_agent(payload, context):

この @app.entrypoint が Runtime 側の公開インターフェースになります。

今の payload は単純で、prompt キーだけを見ています。

prompt = ""
if isinstance(payload, dict):
    prompt = str(payload.get("prompt", "")).strip()

空文字ならエラーとして返し、値があれば Strands Agent を呼びます。

response = agent(prompt)
text = extract_text(response)
return {
    "appName": get_app_name(),
    "message": text,
}

ここで appName と message を返しているので、Lambda2 側では message を読むだけで返信文として扱えます。

また、ログでは以下のように session ID と request ID だけを出しています。

logger.info(
    "agentcore_invocation session_id=%s request_id=%s has_prompt=%s",
    getattr(context, "runtime_session_id", "unknown"),
    getattr(context, "request_id", "unknown"),
    True,
)

ユーザー本文はログに出していません。公開環境では、この判断はかなり重要です。

セッション設計

会話継続のために、AgentCore Runtime の runtimeSessionId には LINE userId を使っています。

ただし、実装では以下のように namespace を足せるようにしています。

<LINE userId>:v2

これは、デプロイ直後に古い Runtime セッションが残っているケースを切り分けるためです。

実際、今回の実装でも以下のようなことが起きました。

  • Runtime version 2 に更新されている
  • DEFAULT endpoint も version 2 を向いている
  • それでも古い model ID を使ったエラーが出る

原因は、同じ runtimeSessionId に対して古いセッションが残っていたためでした。AGENTCORE_SESSION_NAMESPACE を付けることで、新しいセッションとして切り替えて解決できました。

IAM 周りでハマった点

この構成では、IAM の落とし穴がいくつかありました。

AgentCore Runtime 実行ロールの ECR 権限

CDK asset 用 ECR からイメージを pull するために、以下が必要でした。

  • ecr:GetAuthorizationToken
  • ecr:BatchGetImage
  • ecr:GetDownloadUrlForLayer

最初、GetAuthorizationToken を落としていて Runtime 作成に失敗しました。

Lambda2 の InvokeAgentRuntime 権限

bedrock-agentcore:InvokeAgentRuntime を runtime ARN に対して付けただけでは足りませんでした。

実際には、呼び出し先が runtime-endpoint/DEFAULT になっていたため、以下のように endpoint 側も許可する必要がありました。

  • arn:aws:bedrock-agentcore:...:runtime/<runtime-id>
  • arn:aws:bedrock-agentcore:...:runtime/<runtime-id>/runtime-endpoint/*

この部分が抜けると AccessDeniedException になります。

ハマったポイントまとめ

今回ハマったポイントをまとめます。

Python 3.9 で依存が入らない

strands-agents と bedrock-agentcore が Python 3.10 以上が前提なので、Python 3.9 の仮想環境ではインストールできませんでした。（前にも同じハマり方した...）

AgentCore Runtime の ECR pull 権限

CDK asset ECR から Runtime が pull できず、Runtime 作成に失敗しました。

InvokeAgentRuntime の AccessDenied

runtime endpoint 側の ARN を許可していなかったため、Lambda2 から Runtime が呼べませんでした。

CDKでAgentCore Runtimeをデプロイした後にLambda2を上書きするように記述した方がよかったです。

デプロイ後も古いセッションを踏む

Runtime version を更新しても、同じ runtimeSessionId を使っていると古い session に当たり続けることがありました。

セキュリティ上の注意

この種の構成では、LINE 側の機密情報がリポジトリに残りやすいです。

特に以下は Git に commit しないよう注意してください。

  • .env
  • 秘密鍵, 公開鍵
  • チャンネルシークレット
  • チャンネルアクセストークン

また、以下も不要なら commit しないほうがよいです。

  • cdk.out/
  • .venv/
  • tmp-home/

もし漏洩した可能性がある場合は、次を実施してください。

  1. チャンネルシークレット を再発行する
  2. チャンネルアクセストークン を 再発行する
  3. Secrets Manager を更新する
  4. 再デプロイする

17. まとめ

今回は、LINE を入口にして Bedrock AgentCore Runtime 上の Strands Agent を呼び出す最小構成を作成しました。

構成としてはかなりシンプルですが、以下のような拡張余地があります。

  • AgentCore Memoryを使って会話履歴を参照する
  • 家計簿や予定管理用の tool や AgentSkill を追加する
  • structured output でアクションを分岐する

個人的に、「常時起動の自律エージェントを持ちたいが、まずはサーバレスで小さく始めたい」というニーズを持っていたので、これを基に生活基盤を支えるエージェントに育てていきたいと思っています。

また、普段から使っている LINE をフロントにすると、専用のWeb UIを作らずにすぐ会話できるのがかなり楽でした。あとは部屋に設置しているAlexaからも同じエージェントを呼び出せるようにしたいなと思っています。

まだまだ改善の余地はありますが、まずは Bedrock AgentCore Runtime を使って何か1つ動くものを作りたい方の参考になれば嬉しいです。

7
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
7
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?