More than 1 year has passed since last update.

【解決！！！】Bedrock AgentCoreからLangfuseへトレースを送信する！

Last updated at 2025-09-28Posted at 2025-09-23

はじめに(謝罪)

Bedrockerな皆さん、Strands Agents SDK×Bedrock AgentCoreでのエージェント開発、進んでいらっしゃいますでしょうか。
私は進んでません()。

なんだかエージェントが思ったように動いてくれていない、CloudWatch Logsで見ると同じツールが3〜4回に渡って実行され、それのせいでトークンが爆発している…ように見えるような感じの状態になっており大変困ってます。

なのでLangfuseでもう少しわかりやすく実行結果を纏めてみることにしたのですが、その情報がなかなか世に無い…という状況だったため自分で試してみました。

~~が、できませんでした。解決策ご存じの方いらっしゃればご教示いただけますと幸いです…~~

【9/27追記】
有識者の方に助けていただき、トレース送信できるようになりました！！
@neruneruo さん、 @moritalous さん、お力添えいただきありがとうございました！

以下を参考にして、一部アレンジしたものをGitHubで公開しております。

1. まずはローカルのStrands AgentsとLangfuseを統合する

以下公式のサンプルを参考に、実装してみました。

langfuse_agent.py

import os
import base64
 
# Langfuseのパブリックキー・シークレットキーを環境変数に設定
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-******************"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-******************"
# host = "https://cloud.langfuse.com" # EU region
host = "https://us.cloud.langfuse.com" # US region

# Basic Auth headerを構築
LANGFUSE_AUTH = base64.b64encode(
    f"{os.environ.get('LANGFUSE_PUBLIC_KEY')}:{os.environ.get('LANGFUSE_SECRET_KEY')}".encode()
).decode()
 
# OpenTelemetryエンドポイント・ヘッダーを設定
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = host + "/api/public/otel"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

from strands import Agent
from strands.telemetry import StrandsTelemetry
from strands.models.bedrock import BedrockModel
from strands_tools.current_time import current_time
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client

# エージェントのシステムプロンプト
system_prompt = """
あなたは優秀な情報検索エージェントです。
2つのツールを使ってユーザーからの質問を検索・調査し、その内容を簡潔にまとめてください。
回答する際は必ず調査結果をベースとしてください。あなた自身の知識を元に回答することは禁止です。
"""
 
# エージェントのモデル
model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0", # Example model ID
)

# MCPクライアントの設定
streamable_http_mcp_client = MCPClient(lambda: streamablehttp_client("https://mcp.tavily.com/mcp/?tavilyApiKey=<your-api-key>"))
 
# テレメトリーの設定
strands_telemetry = StrandsTelemetry().setup_otlp_exporter()

# MCPセッション開始
with streamable_http_mcp_client:
    # MCPツールを取得後、current_timeツールを追加してエージェントに渡す
    mcp_tools = streamable_http_mcp_client.list_tools_sync()
    agent_tools = mcp_tools + [current_time]

    # エージェントの設定
    agent = Agent(
        model=model,
        tools=agent_tools,
        system_prompt=system_prompt,
        trace_attributes={
            "session.id": "abc-1234", # Example session ID
            "user.id": "user-email-example@domain.com", # Example user ID
            "langfuse.tags": [
                "Strands-Agent-SDK-Example",
                "Strands-Project-Demo",
                "Observability-Tutorial"
            ]
        }
    )

    results = agent("今日から何日後に今年のre:Inventって開催される？")

実行結果は以下のようになりました。

ターミナル

$ uv run langfuse_agent.py
今年のAWS re:Inventがいつ開催されるか調べますね。現在の日付と開催日を確認して何日後かを計算します。

まず現在の日付を確認します。
Tool #1: current_time
次に今年のAWS re:Inventの開催日について検索します。
Tool #2: tavily_search
検索結果から、今年のAWS re:Inventの開催日について明確な情報が得られませんでした。もう少し具体的に検索してみます。
Tool #3: tavily_search
これで情報が得られました。今年(2025年)のAWS re:Inventは2025年12月1日から12月5日まで開催されることがわかりました。

現在日付は2025年9月23日ですので、12月1日の開始日までの日数を計算します。

2025年9月23日から2025年12月1日までは:
- 9月の残り日数: 7日 (23日から30日まで)
- 10月の日数: 31日
- 11月の日数: 30日
- 12月: 1日

したがって、7 + 31 + 30 + 1 = 69日後に今年のAWS re:Inventが開催されます。

今日から69日後の2025年12月1日から12月5日まで、ラスベガスのベネチアンホテルで今年のAWS re:Inventが開催されます。%

という結果になりました。内容自体は成功(日付の計算は諸説、68日が正？)ですが、1度目では情報の取得に失敗しているようです。
何があったのか、Langfuseで見てみます。

query自体は「2025年 AWS re:Invent 開催日日程」と正しそうです。

検索結果にも「今年の12/1~12/5に」と含まれていますが、ここではこれが開催日だと認識されなかったようです。

再度実行し、今度は無事日程を取得できたようでした。

ということで、「動作自体は問題ないが検索のかけ方に改善の余地あり」ということがわかりました。
ただこれはターミナルでの結果だけをみているとブラックボックスなので、この可視性はかなり重要だということがわかりますね…

何気に、Agent Loopが可視化されるのめっちゃわかりやすくないですか？
公式のGenAI Observabilityにもこのくらい期待したいところです。

機密情報をきちんと隠す

次はデプロイなのですが、現状だと機密情報がそのままになっています。Langfuseのキー2つおよびTavily APIキーをベタ書きにしたままデプロイするのは頷けません。

ということで、以下記事を参考にBedrock AgentCore Identityからサクッと3つのキーを取得してきましょう。

AgentCore IdentityにAPIキーを設定する

以下のようにして、Langfuseのパブリックキー・シークレットキーとTavily APIキーを格納します。

ここでの名前を、後続のコード修正で用います。
(ここではそれぞれlangfuse_public_key, langfuse_secrets_key, tavily-api-keyを設定したものとして進めます。)

コード修正

先ほどベタ書きしていた部分を置き換えていきましょう。

AgentCore Identityから値を取得する際には、@requires_api_key()メソッドが最高です。

サンプル

import os
import base64
from bedrock_agentcore.identity.auth import requires_api_key

@requires_api_key(provider_name="langfuse_public_key")
def need_public_api_key(*, api_key: str):
    print("Langfuseパブリックキーを取得")
    os.environ["LANGFUSE_PUBLIC_KEY"] = api_key

@requires_api_key(provider_name="langfuse_secret_key")
def need_secret_api_key(*, api_key: str):
    print("Langfuseシークレットキーを取得")
    os.environ["LANGFUSE_SECRET_KEY"] = api_key

@requires_api_key(provider_name="tavily-api-key")
def need_tavily_api_key(*, api_key: str):
    print("Tavily APIキーを取得")
    os.environ["TAVILY_API_KEY"] = api_key
 
# Langfuseのパブリックキーを取得
if not os.environ.get("LANGFUSE_PUBLIC_KEY"):
    need_public_api_key(api_key="")

# Langfuseのシークレットキーを取得
if not os.environ.get("LANGFUSE_SECRET_KEY"):
    need_secret_api_key(api_key="")

# TavilyのAPIキーを取得
if not os.environ.get("TAVILY_API_KEY"):
    need_tavily_api_key(api_key="")

Tavily APIキーを設定する箇所もこんな感じで修正します。

langfuse_agent.py

# MCPクライアントの設定
- streamable_http_mcp_client = MCPClient(lambda: streamablehttp_client("https://mcp.tavily.com/mcp/?tavilyApiKey=<your-api-key>"))
+ streamable_http_mcp_client = MCPClient(lambda: streamablehttp_client(f"https://mcp.tavily.com/mcp/?tavilyApiKey={os.environ.get('TAVILY_API_KEY')}"))

ローカルでテストすると、先程同様きちんと動くはずです。

2. AgentCore Runtimeへデプロイする

Bedrock AgentCore Starter Toolkitを使って、デプロイやっていきましょう。

コード修正

ただ、デプロイする際はさっきのコードでは動きません。
以下の感じでコードを修正する必要があります。

langfuse_agent.py

import os
import json
import base64
import logging
import boto3

from strands import Agent
from strands.telemetry import StrandsTelemetry 
from strands.models.bedrock import BedrockModel
from strands_tools.current_time import current_time
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client
from bedrock_agentcore import BedrockAgentCoreApp
from bedrock_agentcore.identity.auth import requires_api_key
from langfuse import get_client

logger = logging.getLogger("langfuse_agent")
logging.basicConfig(level=logging.INFO)

langfuse_public_key_public_id = os.environ["LANGFUSE_PUBLIC_KEY_SECRET_ID"]
langfuse_public_key_secret_id = os.environ["LANGFUSE_SECRET_KEY_SECRET_ID"]

# Secrets ManagerからLangfuseのキーを取得
# 期待するデータ格納形式:
# - SecretId: "langfuse-public-key" → JSON形式 {"api_key_value":"実際のキー値"}
# - SecretId: "langfuse-secret-key" → JSON形式 {"api_key_value":"実際のキー値"}
secrets_manager = boto3.client("secretsmanager", region_name="us-east-1")

# Langfuse Public Keyを取得（JSON形式 {"api_key_value":"実際のキー値"} から取得）
public_secret = secrets_manager.get_secret_value(SecretId=langfuse_public_key_public_id)
public_data = json.loads(public_secret["SecretString"])
os.environ["LANGFUSE_PUBLIC_KEY"] = public_data["api_key_value"]
print(f"Langfuse パブリックキーを取得: {public_data['api_key_value'][:4]}...{public_data['api_key_value'][-4:]}:")

# Langfuse Secret Keyを取得（JSON形式 {"api_key_value":"実際のキー値"} から取得）
secret_key_secret = secrets_manager.get_secret_value(SecretId=langfuse_public_key_secret_id)
secret_data = json.loads(secret_key_secret["SecretString"])
os.environ["LANGFUSE_SECRET_KEY"] = secret_data["api_key_value"]
print(f"Langfuse シークレットキーを取得: {secret_data['api_key_value'][:4]}...{secret_data['api_key_value'][-4:]}:")

# Tavily APIキーはAgentCore Identityから取得
@requires_api_key(provider_name="tavily-api-key")
def need_tavily_api_key(*, api_key: str):
    print(f"Tavily APIキーを取得: {api_key[:4]}...{api_key[-4:]}")
    os.environ["TAVILY_API_KEY"] = api_key

LANGFUSE_AUTH = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    os.environ.get("LANGFUSE_HOST", "https://us.cloud.langfuse.com") + "/api/public/otel"
)

os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

# os.environ["LANGFUSE_DEBUG"] = "True"

os.environ["OTEL_TRACES_EXPORTER"] = "otlp"

os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = "http/protobuf"

strands_telemetry = StrandsTelemetry().setup_otlp_exporter()
langfuse = get_client()

# エージェントのシステムプロンプト
system_prompt = """
あなたは優秀な情報検索エージェントです。
2つのツールを使ってユーザーからの質問を検索・調査し、その内容を簡潔にまとめてください。
回答する際は必ず調査結果をベースとしてください。あなた自身の知識を元に回答することは禁止です。
"""
 
# エージェントのモデル
model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
)

app = BedrockAgentCoreApp()

@app.entrypoint
async def invoke_agent(payload: dict):
    user_message = payload.get("prompt", "今日から数えて、今年のre:Inventって何日後？")
    
    # ワークロードアクセストークン用の設定
    os.environ.setdefault("WORKLOAD_NAME", "langfuse-agent")
    os.environ.setdefault("USER_ID", "langfuse")

    # AgentCore IdentityからTavily APIキーを取得
    if not os.environ.get("TAVILY_API_KEY"):
        need_tavily_api_key(api_key="")

    # エージェントをビルドし、実行
    logger.info("[invoke] received prompt (len=%d)", len(user_message))
    tavily_api_key = os.environ.get("TAVILY_API_KEY")
    if not tavily_api_key:
        raise RuntimeError("TAVILY_API_KEY が未取得です。")

    # MCPクライアントの設定
    streamable_http_mcp_client = MCPClient(lambda: streamablehttp_client(f"https://mcp.tavily.com/mcp/?tavilyApiKey={tavily_api_key}"))

    # MCPセッション開始
    with streamable_http_mcp_client:
        # MCPツールを取得後、current_timeツールを追加してエージェントに渡す
        mcp_tools = streamable_http_mcp_client.list_tools_sync()
        agent_tools = mcp_tools + [current_time]

        # エージェントの設定
        agent = Agent(
            model=model,
            tools=agent_tools,
            system_prompt=system_prompt,
            callback_handler=None,
        )
        output = agent(user_message)

        langfuse.flush()

        logger.info("[invoke] done")
        return {"output": output}

if __name__ == "__main__":
    app.run()

修正点1. AWS SDKでSecrets Managerからキーを取得

ローカルで試した際には、AgentCore Identity(裏側はSecrets Managerですが)からLangfuseのキーを取得していました。
が、AWS SDKを用いてシンプルにSecrets Managerから取得する方がシンプルで良いと思います。

また、AgentCore Identityから値を取得する際には、必ず @app.entrypoint でデコレートした関数内で実行しないといけないのですが、それが厄介そうだったので、こちらの方式に修正しました。

上記コード内では {"api_key_value":"Langfuseのキー"} の形式で予めシークレットを格納しておき、そのシークレット名を環境変数に設定しておくことで、Langfuseのキーを取り出すことができるようになっています。

langfuse_agent.py

langfuse_public_key_public_id = os.environ["LANGFUSE_PUBLIC_KEY_SECRET_ID"]
langfuse_public_key_secret_id = os.environ["LANGFUSE_SECRET_KEY_SECRET_ID"]

# Secrets ManagerからLangfuseのキーを取得
# 期待するデータ格納形式:
# - SecretId: "langfuse-public-key" → JSON形式 {"api_key_value":"実際のキー値"}
# - SecretId: "langfuse-secret-key" → JSON形式 {"api_key_value":"実際のキー値"}
secrets_manager = boto3.client("secretsmanager", region_name="us-east-1")

# Langfuse Public Keyを取得（JSON形式 {"api_key_value":"実際のキー値"} から取得）
public_secret = secrets_manager.get_secret_value(SecretId=langfuse_public_key_public_id)
public_data = json.loads(public_secret["SecretString"])
os.environ["LANGFUSE_PUBLIC_KEY"] = public_data["api_key_value"]
print(f"Langfuse パブリックキーを取得: {public_data['api_key_value'][:4]}...{public_data['api_key_value'][-4:]}:")

# Langfuse Secret Keyを取得（JSON形式 {"api_key_value":"実際のキー値"} から取得）
secret_key_secret = secrets_manager.get_secret_value(SecretId=langfuse_public_key_secret_id)
secret_data = json.loads(secret_key_secret["SecretString"])
os.environ["LANGFUSE_SECRET_KEY"] = secret_data["api_key_value"]
print(f"Langfuse シークレットキーを取得: {secret_data['api_key_value'][:4]}...{secret_data['api_key_value'][-4:]}:")

AgentCoreサービスロールにおいて、Secrets Managerからシークレットを取得するための権限を追加しておいてください。

AgentCoreサービスロールに以下権限を追加

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "secretsmanager:GetResourcePolicy",
                "secretsmanager:GetSecretValue",
                "secretsmanager:DescribeSecret",
                "secretsmanager:ListSecretVersionIds"
            ],
            "Resource": [
                "<Langfuseパブリックキーを設定したシークレットのARN>",
                "<Langfuseシークレットキーを設定したシークレットのARN>"
            ]
        }
    ]
}

修正点2. Langfuseへのトレース送信設定を追加

以下コードの部分であり、Bedrockerのお二人に一番助けていただいた部分です。

langfuse_agent.py

LANGFUSE_AUTH = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    os.environ.get("LANGFUSE_HOST", "https://us.cloud.langfuse.com") + "/api/public/otel"
)

os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

# os.environ["LANGFUSE_DEBUG"] = "True"

os.environ["OTEL_TRACES_EXPORTER"] = "otlp"

os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = "http/protobuf"

strands_telemetry = StrandsTelemetry().setup_otlp_exporter()
langfuse = get_client()

LangfuseはOpenTelemetry Protocol(OTLP)を用いてトレースを受信しています。

ざっくり解説すると、環境変数を使用してOTLPエクスポーターの設定を行っていき、StrandsTelemetryでその設定を初期化する、という流れになります。
Telemetryを初期化する際に環境変数を自動で読み込んでくれるというのがポイントです。

OTLPエクスポーターにおいて、ここでは大きく3つの設定を行っています。

OTLPエンドポイント

os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = (
    os.environ.get("LANGFUSE_HOST", "https://us.cloud.langfuse.com") + "/api/public/otel"
)

ここにLangfuseのエンドポイントを設定します
設定例は https://(us.)cloud.langfuse.com/api/public/otel
設定したエンドポイントにトレースやメトリクス、ログなどの情報を送信します

OTLPヘッダー

LANGFUSE_AUTH = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()

os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

エンドポイントへのリクエストに追加するヘッダー
base64でエンコードされたAPIキーを設定する
Langfuseはリクエストの認証にBasic Authを使用

OTLPプロトコル

os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = "http/protobuf"

データ送信時に用いるプロトコルを設定します
以下の値を設定できますが、Langfuseでは http/protobuf のみ設定可能です
- grpc
- http/protobuf
- http/json
ただし、agentcore configureコマンド実行時に作成されるDockerfileにおいては、gRPCがデフォルトで用いられるようです
したがって、Dockerfileを以下のように修正したうえでデプロイする必要があります

Dockerfile

- CMD ["opentelemetry-instrument", "python", "-m", "langfuse_agent"]
+ CMD ["python", "-m", "langfuse_agent"]

opentelemetry-instrumentはラッパー実行コマンドで、これがあるとOtelの自動計測を初期化し、トレーサーやエクスポーターを設定します。
これらがあるおかげで、普段特別意識/設定せずともGenAI Observabilityにトレースが送信されています。

ただ、これはデフォルトでgRPCを用いるため、削除しておく必要がありそうです。

また、削除理由としてはOtelの初期化が重複してしまうからというのもあります。
アプリコード内でOtelの初期化をしていますが、本来はopentelemetry-instrumentで初期化しています。

後から無理やり修正すると、WARNING:opentelemetry.instrumentation.instrumentor:Attempting to instrument while already instrumentedという警告が出てくるので、一旦除外しておかないといけないのかなと思います。

トレース送信設定初期化

実はどちらか1つを設定すればOKみたいです。

strands_telemetry = StrandsTelemetry().setup_otlp_exporter()

StrandsAgents用のOpenTelemetryセットアップ用処理で、OpenTelemetry経由での送信となります
内部でOpenTemeletry SDKを用いて以下の設定を行います
- OTLPSpanExporter(...) を作成 (HTTP or gRPCは環境変数で決まる)
- BatchSpanProcessor(...) を作ってTracerProviderに登録
最終的にStrands側のspanを、上記で設定したOTLPエンドポイントへ送信
この設定を行う前に、OTLPの環境変数を設定しておく必要がある

strands_telemetry = StrandsTelemetry().setup_otlp_exporter()
langfuse = get_client()

LangfuseのPython SDKを初期化します
環境変数(LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY、LANGFUSE_HOST)から設定を読んでいる
- SDK独自のイベント／トレース API（client.trace(...) など）を使えるようになる
- 送信をバックグラウンドでバッチし、client.flush()で強制送信できる
ただ現在の実装だとOpenTelemetry経路でのトレース送信が有効になっているのみなので、不要かもしれません
- セッション管理・スコア・任意メタ情報も送信したい場合はこちらを用いるようです

まとめ

アプリ実装側での環境変数設定とDockerfile修正によって、AgentCoreからでLangfuseへのトレース送信ができるようになりました！

これでエージェント開発から運用まで、加速させてやっていきましょう！

参考資料

より詳しく知りたい方は、OTLPエクスポーターとLangfuseの設定ドキュメントを呼んでみると、中身がよく分かると思います。

2. AgentCore Runtimeへデプロイする【9/27更新、こちらは読まなくてOKです】

自身の美貌として残していますが、参照不要です

Bedrock AgentCore Starter Toolkitを使って、デプロイやっていきましょう。

https://github.com/aws/bedrock-agentcore-starter-toolkit

コード修正

ただ、デプロイする際はさっきのコードでは動きません。
以下の感じでコードを修正する必要があります。

langfuse_agent.py

import os
import base64
import logging

from strands import Agent
from strands.models.bedrock import BedrockModel
from strands_tools.current_time import current_time
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client
from bedrock_agentcore import BedrockAgentCoreApp
from bedrock_agentcore.identity.auth import requires_api_key

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor, ConsoleSpanExporter
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from bedrock_agentcore.identity.auth import requires_api_key

logger = logging.getLogger("langfuse_agent")
logging.basicConfig(level=logging.DEBUG)

# OpenTelemetryログを取りたい(デバッグログ)
logging.getLogger("opentelemetry.exporter.otlp.proto.http.trace_exporter").setLevel(logging.DEBUG)
logging.getLogger("opentelemetry").setLevel(logging.DEBUG)

_langfuse_attached: bool = False

# provider_nameには、AgentCore Identityで設定した名前を設定
@requires_api_key(provider_name="langfuse_public_key")
def need_public_api_key(*, api_key: str):
    print(f"Langfuseパブリックキーを取得: {api_key[:4]}...{api_key[-4:]}")
    os.environ["LANGFUSE_PUBLIC_KEY"] = api_key

@requires_api_key(provider_name="langfuse_secret_key")
def need_secret_api_key(*, api_key: str):
    print(f"Langfuseシークレットキーを取得: {api_key[:4]}...{api_key[-4:]}")
    os.environ["LANGFUSE_SECRET_KEY"] = api_key

@requires_api_key(provider_name="tavily-api-key")
def need_tavily_api_key(*, api_key: str):
    print(f"Tavily APIキーを取得: {api_key[:4]}...{api_key[-4:]}")
    os.environ["TAVILY_API_KEY"] = api_key

def _ensure_sdk_provider() -> SDKTracerProvider:
    provider = trace.get_tracer_provider()
    if isinstance(provider, SDKTracerProvider):
        return provider  # type: ignore[return-value]
    # 念のためのフォールバック（通常は不要）
    sdk = SDKTracerProvider(resource=Resource.create({
        "service.name": os.getenv("OTEL_SERVICE_NAME", "strands-langfuse-demo")
    }))
    trace.set_tracer_provider(sdk)
    return sdk

def _attach_langfuse_exporter_once():
    global _langfuse_attached
    if _langfuse_attached:
        return

    if "LANGFUSE_PUBLIC_KEY" not in os.environ: need_public_api_key(api_key="")
    if "LANGFUSE_SECRET_KEY" not in os.environ: need_secret_api_key(api_key="")

    auth = base64.b64encode(
        f'{os.environ["LANGFUSE_PUBLIC_KEY"]}:{os.environ["LANGFUSE_SECRET_KEY"]}'.encode()
    ).decode()

    base = os.environ.get("LANGFUSE_BASE_URL", "https://us.cloud.langfuse.com")
    endpoint = f"{base}/api/public/otel/v1/traces"

    provider = _ensure_sdk_provider()
    exporter = OTLPSpanExporter(
        endpoint=endpoint,
        headers={"Authorization": f"Basic {auth}"},
        timeout=10,
    )
    provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
    provider.add_span_processor(BatchSpanProcessor(exporter, max_export_batch_size=64, schedule_delay_millis=500))
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    _langfuse_attached = True
    logger.info("[OTEL] Langfuse exporter attached endpoint=%s", endpoint)


# エージェントのシステムプロンプト
system_prompt = """
あなたは優秀な情報検索エージェントです。
2つのツールを使ってユーザーからの質問を検索・調査し、その内容を簡潔にまとめてください。
回答する際は必ず調査結果をベースとしてください。あなた自身の知識を元に回答することは禁止です。
"""
 
# エージェントのモデル
model = BedrockModel(
    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
)

app = BedrockAgentCoreApp()

@app.entrypoint
async def invoke_agent(payload: dict):
    prompt = (payload or {}).get("prompt", "今日から数えて、今年のre:Inventって何日後？")
    if not isinstance(prompt, str) or not prompt.strip():
        logger.warning("[invoke] missing prompt")
        return {"error": "prompt is required"}
    
    # ワークロードアクセストークン用の設定
    os.environ.setdefault("WORKLOAD_NAME", "langfuse-agent")
    os.environ.setdefault("USER_ID", "langfuse")

    # AgentCore IdentityからTavily APIキーを取得

    if not os.environ.get("TAVILY_API_KEY"):
        need_tavily_api_key(api_key="")
    
    _attach_langfuse_exporter_once()

    # ======= デバッグ =======
    tracer = trace.get_tracer("langfuse.selftest")
    span = tracer.start_span("langfuse.selftest.probe")
    logger.info("[OTEL] probe: is_recording=%s provider=%s",
    getattr(span, "is_recording", lambda: None)(), type(trace.get_tracer_provider()))
    span.end()
    with tracer.start_as_current_span("langfuse.selftest") as s:
        if hasattr(s, "is_recording") and not s.is_recording():
            logger.warning("[OTEL] selftest span is NOT recording")
    # ======= デバッグ =======

    # エージェントをビルドし、実行
    logger.info("[invoke] received prompt (len=%d)", len(prompt))
    tavily_api_key = os.environ.get("TAVILY_API_KEY")
    if not tavily_api_key:
        raise RuntimeError("TAVILY_API_KEY が未取得です。")

    # MCPクライアントの設定
    streamable_http_mcp_client = MCPClient(lambda: streamablehttp_client(f"https://mcp.tavily.com/mcp/?tavilyApiKey={tavily_api_key}"))

    # MCPセッション開始
    with streamable_http_mcp_client:
        # MCPツールを取得後、current_timeツールを追加してエージェントに渡す
        mcp_tools = streamable_http_mcp_client.list_tools_sync()
        agent_tools = mcp_tools + [current_time]

        # エージェントの設定
        agent = Agent(
            model=model,
            tools=agent_tools,
            system_prompt=system_prompt,
            trace_attributes={
                "session.id": "abc-1234", # Example session ID
                "user.id": "user-email-example@domain.com", # Example user ID
                "langfuse.tags": [
                    "Strands-Agent-SDK-Example",
                    "Strands-Project-Demo",
                    "Observability-Tutorial"
                ]
            }
        )
        output = agent(prompt.strip())

        # ======= デバッグ =======
        tracer = trace.get_tracer("langfuse.selftest")
        with tracer.start_as_current_span("langfuse.selftest"):
            pass
        # レスポンス返却前に flush（BatchSpanProcessor 依存だと遅延することがあるため）
        provider = trace.get_tracer_provider()
        force_flush = getattr(provider, "force_flush", None)
        if callable(force_flush):
            try:
                ok = force_flush(10000)  # ms
                logger.info("[OTEL] force_flush => %s", ok)
            except Exception as e:
                logger.warning("[OTEL] force_flush failed: %s", e)
        # ======= デバッグ =======
        
        logger.info("[invoke] done")
        return {"output": output}

if __name__ == "__main__":
    app.run()

修正点1. `@requires_api_key`はentrypoint内で実行

先程まではこの関数をモジュールimport時に呼び出していましたが、デプロイ後はRuntimeがリクエスト時に渡すWorkload Access Tokenを使うため失敗するようになります。

なので、@app.entrypoint付きの関数内でコールします。

@app.entrypoint
async def invoke_agent(payload: dict):
    # 中略
    # ↓ここの中でLangfuseのキーを取得する
    _attach_langfuse_exporter_once()

修正点2. LangfuseへのOTLP/HTTP Exporterを後付け

_attach_langfuse_exporter_once()で以下を実施しています。

SDK Provider を確保（必要ならフォールバック生成）
endpoint=https://.cloud.langfuse.com/api/public/otel/v1/traces を設定
headers={"Authorization": "Basic "} を設定

を設定し、Batch+SimpleのProcessorをアタッチしておく(ConsoleSpanExporterは診断用）。

修正点3. `invoke_agent()`を追加

先程までに作成した関数を利用して必要な設定を行い、最終的にエージェントを実行します。

@app.entrypoint
async def invoke_agent(payload: dict):
    # Telemetry初期化
    # Tavily MCPのツールをロードして Agent を構築・実行

if __name__ == "__main__":
    app.run()

デプロイ

ようやくデプロイです。
まずは以下のようなフォルダ構成にします。

root/
  __init__.py (空でOK)
  langfuse_agent.py
  requirements.txt

requirements.txtは以下の感じです。

requirements.txt

# Bedrock AgentCore SDK
bedrock-agentcore>=0.1.0

# Strands Agents SDK
strands-agents>=1.0.0
strands-agents-tools>=0.1.5

opentelemetry-exporter-otlp-proto-http>=1.26.0
opentelemetry-sdk>=1.20.0

ターミナルでスターターツールキットをインストールしておきます(以下はuvを使っている場合)。

ターミナル

$ uv pip install bedrock-agentcore-starter-toolkit

agentcore configureとDockerfile修正

agentcoreコマンドが使えるようになったので、デプロイです。

ターミナル

$ agentcore configure --entrypoint langfuse_agent.py
# IAMロール、ECRリポジトリは任意。その後はenterでOK

ここで作成されたDockerfileにおいて、以下のように修正します。
Runtime既定のAWS Distro for Open Telemetry (ADOT) SDKを用いない手法に変更しています。

Dockerfile

FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim

ENV UV_SYSTEM_PYTHON=1 UV_COMPILE_BYTECODE=1 PYTHONUNBUFFERED=1 \
     AWS_REGION=us-east-1 AWS_DEFAULT_REGION=us-east-1 \
     DOCKER_CONTAINER=1 \
+    DISABLE_ADOT_OBSERVABILITY=true \
+    OTEL_PYTHON_DISTRO=aws_distro \
+    OTEL_PYTHON_CONFIGURATOR=aws_configurator \
+    OTEL_SERVICE_NAME=strands-langfuse-demo \
+    OTEL_PROPAGATORS=tracecontext,baggage,xray \
+    OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf \
+    OTEL_TRACES_SAMPLER=always_on

+ RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates \
+ && rm -rf /var/lib/apt/lists/*

COPY requirements.txt requirements.txt

- RUN uv pip install -r requirements.txt && \
-     uv pip install aws-opentelemetry-distro>=0.10.1
+ RUN uv pip install -r requirements.txt && \
+     uv pip install "aws-opentelemetry-distro>=0.10.1"

EXPOSE 8080 8000

COPY . .

# Use the full module path
- CMD ["opentelemetry-instrument", "python", "-m", "langfuse_agent"]
+ CMD ["python", "-m", "langfuse_agent"]

aws-opentelemetry-distroは、AWS向けにチューニングされた OpenTelemetry（OTel）のディストリビューションだそうです。X-Ray/CloudWatchと親和性があり、これでGenAI Observabilityが動いているとのこと。

ただ、これがあるとLangfuseには送信できません。

理由1つ目としては、Otelの初期化が重複してしまうから。
アプリコード内でOtelの初期化をしていますが、本来はopentelemetry-instrumentで初期化しています。

理由2つ目としては、LangfuseがgRPCというOtelのエンドポイントをサポートしていないため。
opentelemetry-instrumentationのデフォルトエクスポート先はOTLP gRPCのため、これを変更して上げる必要があります。

https://langfuse.com/integrations/native/opentelemetry

https://pypi.org/project/opentelemetry-instrumentation/

なのでここではできるだけシンプルに設定を変更する目的で、opentelemetry-instrumentationを用いない方法に変えています。

$ agentcore launch

# 動作確認
$ agentcore invoke '{"prompt": "今日の日付は？"}' --user-id "langfuse"

これできちんとデプロイされ、エージェントは動くはずです。

が、なんだかちゃんとトレース送信できません。
DEBUG:urllib3.connectionpool:https://us.cloud.langfuse.com:443 "POST /api/public/otel/v1/traces HTTP/1.1" 200 Noneというのは返ってくるんですが、私の環境だとLangfuse上にこの実行結果が表示されません。

以下ドキュメントではLangfuseに対応しているとは記載されていないため、もしかすると対応待ちなのかもしれませんが…できる方法をご存じの方いらっしゃれば共有いただけますと幸いです。

https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/observability-configure.html

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

【解決！！！】Bedrock AgentCoreからLangfuseへトレースを送信する！

はじめに(謝罪)

1. まずはローカルのStrands AgentsとLangfuseを統合する

機密情報をきちんと隠す

AgentCore IdentityにAPIキーを設定する

コード修正

2. AgentCore Runtimeへデプロイする

コード修正

修正点1. AWS SDKでSecrets Managerからキーを取得

修正点2. Langfuseへのトレース送信設定を追加

OTLPエンドポイント

OTLPヘッダー

OTLPプロトコル

トレース送信設定初期化

まとめ

参考資料

2. AgentCore Runtimeへデプロイする 【9/27更新、こちらは読まなくてOKです】

コード修正

修正点1. @requires_api_keyはentrypoint内で実行

修正点2. LangfuseへのOTLP/HTTP Exporterを後付け

修正点3. invoke_agent()を追加

デプロイ

agentcore configureとDockerfile修正

2. AgentCore Runtimeへデプロイする【9/27更新、こちらは読まなくてOKです】

修正点1. `@requires_api_key`はentrypoint内で実行

修正点3. `invoke_agent()`を追加