LoginSignup
9
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

2026年!初アウトプットをしよう
@minorun365(みのるん)inKDDIアジャイル開発センター株式会社

【Bedrock AgentCore】謎のデコレーター @app.async_task

Last updated at Posted at 2026-01-12

AIエージェント版のLambdaこと、Bedrock AgentCoreランタイムでは長時間の非同期タスクを実行することができます。最大8時間！

つまり、アプリケーションの利用者がAIエージェントに仕事を依頼したら、リアルタイムに結果の応答を待つだけではなく、時間かかる仕事を「やっといてね」って任せることができます。

まずはデコレーターなしで普通に使う

どういう風に使うのか、公式ドキュメントを見てみましょう。

Pythonサンプルコードが掲載されています。不要な部分を削って、日本語化したものが以下です。

async.py
import time
import threading
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp

# AgentCore SDKでAPIクライアントを作成
app = BedrockAgentCoreApp()

# 非同期タスクを起動するツール
@tool
def start_async():
    # 非同期タスクの追跡を開始
    task_id = app.add_async_task("ダミー処理")

    # バックグラウンドスレッドでタスクを実行
    def background_work():
        time.sleep(10)  # ダミー処理
        app.complete_async_task(task_id)  # ステータスを完了にする

    threading.Thread(target=background_work, daemon=True).start()
    return "バックグラウンドタスクを開始しました"

# Strandsでエージェントを作成
agent = Agent(tools=[start_async])

# APIサーバーのエントリーポイントを作成
@app.entrypoint
def invoke(payload):
    prompt = payload.get("prompt")
    return agent(prompt)

# APIサーバーを起動
if __name__ == "__main__":
    app.run()

わざわざAgentCoreランタイムにデプロイせずとも、ローカルで試せます。

# バックエンドの起動
aws login
python async.py

別のターミナルでHTTPリクエストを実行して、呼び出してみましょう。

curl -X POST http://localhost:8080/invocations \
    -H "Content-Type: application/json" \
    -d '{"prompt": "非同期タスクを実行して"}'

以下のように、一度回答してから非同期タスクを実行してくれます。便利！

非同期タスクを実行します。

Tool #1: async_task

{
    "timestamp": "2026-01-12T13:15:29.425Z",
    "level": "INFO",
    "message": "Async task started: background_processing (ID: 4680601251139015056)",
    "logger": "bedrock_agentcore.app"
}

{
    "timestamp": "2026-01-12T13:15:30.704Z",
    "level": "INFO",
    "message": "Invocation completed successfully (3.631s)",
    "logger": "bedrock_agentcore.app",
    "requestId": "6d895246-3411-4a07-a50c-70733bee61fa"
}

{
    "timestamp": "2026-01-12T13:15:39.426Z",
    "level": "INFO",
    "message": "Async task completed: background_processing (ID: 4680601251139015056, Duration: 10.00s)",
    "logger": "bedrock_agentcore.app"
}

呼び出しは同期・非同期問わず、同じ InvokeAgentRuntime APIが使えるんですね。簡単です。

謎のデコレーター発見

でも app.add_async_task とか app.complete_async_task(task_id) とか毎回手動操作するの、面倒ですよね。

そう思いながら以下のQiitaを読んでいると…

非同期関数に@app.async_taskデコレータを付与するだけで、自動でトラッキングしてくれるようになります。

マ？ そんなのドキュメントに書いてないけど…

SDKの中身を覗いてみたら、本当にありました！

使ってみましょう。

import asyncio
from bedrock_agentcore.runtime import BedrockAgentCoreApp

# AgentCore SDKでAPIクライアントを作成
app = BedrockAgentCoreApp()

# 非同期タスクを実行する関数
@app.async_task
async def async_task():
    await asyncio.sleep(10)  # ダミー処理
    return "バックグラウンドタスクを完了しました"

# APIサーバーのエントリーポイントを作成
@app.entrypoint
async def handler(event):
    asyncio.create_task(async_task())
    return "バックグラウンドタスクを開始しました"

# APIサーバーを起動
if __name__ == "__main__":
    app.run()

確かに自動でステータス管理してくれます。便利！

{"timestamp": "2026-01-12T13:44:37.999Z", "level": "INFO", "message": "Invocation completed successfully (0.000s)", "logger": "bedrock_agentcore.app", "requestId": "f6ad3876-500e-48ff-b800-f9ac26e5e628"}
{"timestamp": "2026-01-12T13:44:38.000Z", "level": "INFO", "message": "Async task started: async_task (ID: 878873993916010366)", "logger": "bedrock_agentcore.app", "requestId": "f6ad3876-500e-48ff-b800-f9ac26e5e628"}
{"timestamp": "2026-01-12T13:44:48.001Z", "level": "INFO", "message": "Async task completed: async_task (10.002s)", "logger": "bedrock_agentcore.app", "requestId": "f6ad3876-500e-48ff-b800-f9ac26e5e628"}
{"timestamp": "2026-01-12T13:44:48.002Z", "level": "INFO", "message": "Async task completed: async_task (ID: 878873993916010366, Duration: 10.00s)", "logger": "bedrock_agentcore.app", "requestId": "f6ad3876-500e-48ff-b800-f9ac26e5e628"}

ただ、これだと肝心なAIエージェントが含まれていませんね。。

いい感じに使うための試行錯誤

先ほどのサンプルコードみたいに入れてみましょう。

import asyncio
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp

# AgentCore SDKでAPIクライアントを作成
app = BedrockAgentCoreApp()

# 非同期タスクを起動するツール
@tool
@app.async_task
async def start_async(**kwargs):
    await asyncio.sleep(10) # ダミー処理
    return "バックグラウンドタスクが完了しました"

# Strandsでエージェントを作成
agent = Agent(tools=[start_async])

# APIサーバーのエントリーポイントを作成
@app.entrypoint
def invoke(payload):
    prompt = payload.get("prompt")
    return agent(prompt)

# APIサーバーを起動
if __name__ == "__main__":
    app.run()

Strandsの @tool とAgentCoreの @app.async_task デコレーターを併用するため、両者の仕様整合性がとれるよう、関数にキーワード引数を追加しています。

実行すると…おや？先ほどと違って、ランタイム呼び出しの完了が非同期タスクの終了を待ってしまっています。

非同期タスクを実行するために、start_async関数を呼び出します。引数とキーワード引数のパラメータを設定して実行します。
Tool #1: start_async
{"timestamp": "2026-01-12T14:10:35.196Z", "level": "INFO", "message": "Async task started: start_async (ID: 540174483880573374)", "logger": "bedrock_agentcore.app"}
{"timestamp": "2026-01-12T14:10:45.197Z", "level": "INFO", "message": "Async task completed: start_async (10.001s)", "logger": "bedrock_agentcore.app"}
{"timestamp": "2026-01-12T14:10:45.198Z", "level": "INFO", "message": "Async task completed: start_async (ID: 540174483880573374, Duration: 10.00s)", "logger": "bedrock_agentcore.app"}
{"timestamp": "2026-01-12T14:10:47.426Z", "level": "INFO", "message": "Invocation completed successfully (14.937s)", "logger": "bedrock_agentcore.app", "requestId": "3eb6cab9-e2f5-4812-b475-07dd972da272"}

先ほどは threading ライブラリを使ってバックグラウンドタスクを開始していましたが、今回は10秒待つ処理がそのまま invoke 関数を待たせてしまっていますね。

AgentCoreというよりPythonの仕様どおりの動きです。
それでは、非同期処理を別の関数に分けるとどうでしょうか。

import asyncio
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp

# AgentCore SDKでAPIクライアントを作成
app = BedrockAgentCoreApp()

# デコレーターでタスク本体を進捗管理
@app.async_task
async def background_task():
    await asyncio.sleep(10)  # ダミー処理
    return "バックグラウンドタスクが完了しました"

# 非同期タスクを起動するツール
@tool
async def start_async():
    asyncio.create_task(background_task())
    return "バックグラウンドタスクを開始しました"

# Strandsでエージェントを作成
agent = Agent(tools=[start_async])

# APIサーバーのエントリーポイントを作成
@app.entrypoint
def invoke(payload):
    prompt = payload.get("prompt")
    return agent(prompt)

# APIサーバーを起動
if __name__ == "__main__":
    app.run()

今度は不思議なことに、非同期タスクが秒で終了してしまいます。

非同期タスクを開始いたします。
Tool #1: start_async
{"timestamp": "2026-01-12T14:17:35.054Z", "level": "INFO", "message": "Async task started: background_task (ID: -5407902879952980538)", "logger": "bedrock_agentcore.app"}
{"timestamp": "2026-01-12T14:17:36.224Z", "level": "INFO", "message": "Async task completed: background_task (ID: -5407902879952980538, Duration: 1.17s)", "logger": "bedrock_agentcore.app"}
{"timestamp": "2026-01-12T14:17:36.226Z", "level": "INFO", "message": "Invocation completed successfully (4.358s)", "logger": "bedrock_agentcore.app", "requestId": "7a608a7c-a8be-45fd-9026-806f1ec6cf5b"}

どうもStrandsのツール呼び出しが終了すると、その先の非同期タスクも巻き添えを食らってしまうようですね。

最初のサンプルコードのように、Threadingライブラリを使ってみましょう。
Threadingは同期関数しか呼べませんが、@app.async_task デコレーターは非同期関数のみ対応のため、中間関数を作って工夫する必要があります。

import asyncio
import threading
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp

# AgentCore SDKでAPIクライアントを作成
app = BedrockAgentCoreApp()

# デコレーターでタスク本体を進捗管理
@app.async_task
async def background_task():
   await asyncio.sleep(10) # ダミー処理

# バックグラウンドで処理を実行する中間関数
def run_in_background(async_task):
   loop = asyncio.new_event_loop()
   asyncio.set_event_loop(loop)
   try:
       loop.run_until_complete(async_task)
   finally:
       loop.close()

# 非同期処理を起動するツール
@tool
def start_async():
   threading.Thread(
       target=run_in_background,
       args=(background_task(),),
       daemon=True
   ).start()
   return "バックグラウンドタスクを開始しました"

# Strandsでエージェントを作成
agent = Agent(tools=[start_async])

# APIサーバーのエントリーポイントを作成
@app.entrypoint
def invoke(payload):
    prompt = payload.get("prompt")
    return agent(prompt)

# APIサーバーを起動
if __name__ == "__main__":
    app.run()

これなら元と同じように動きました！

非同期タスクを開始します。
Tool #1: start_async
{"timestamp": "2026-01-12T14:38:32.545Z", "level": "INFO", "message": "Async task started: background_task (ID: 1674339398589238552)", "logger": "bedrock_agentcore.app"}
{"timestamp": "2026-01-12T14:38:34.525Z", "level": "INFO", "message": "Invocation completed successfully (3.712s)", "logger": "bedrock_agentcore.app", "requestId": "9c2763b7-d654-44c1-8abc-941d9b201b80"}
{"timestamp": "2026-01-12T14:38:42.545Z", "level": "INFO", "message": "Async task completed: background_task (10.000s)", "logger": "bedrock_agentcore.app"}
{"timestamp": "2026-01-12T14:38:42.546Z", "level": "INFO", "message": "Async task completed: background_task (ID: 1674339398589238552, Duration: 10.00s)", "logger": "bedrock_agentcore.app"}

面倒くさっ！！

これならデコレーター使わなくていいですね。ドキュメントに載ってない理由が分かりました🤣

まとめ

@app.async_task デコレーター、同期関数に対応してくれ〜ｗ

Python得意な人はコントリビュートしてみよう！

9
2
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
9
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?