将 Litefuse 与 Claude Agent SDK 集成

本 notebook 演示如何使用 OpenTelemetry，从 Claude Agent SDK 中捕获详细的 trace 并发送到 Litefuse。

什么是 Claude Agent SDK？
Claude Agent SDK 是 Anthropic 用于构建 AI Agent 的开源框架。它提供简洁的 API 用于创建可调用工具的 Agent，并原生支持 MCP。

什么是 Litefuse？
Litefuse 是开源的 AI Agent 可观测性与评估平台。它为 Agent 中每一个 prompt、模型响应和工具调用提供详细的 tracing、监控和分析，让 LLM 应用的调试、评估和迭代变得简单。

第 1 步：安装依赖

%pip install langfuse claude-agent-sdk "langsmith[claude-agent-sdk]" "langsmith[otel]" -q

第 2 步：设置环境变量

设置你的 Litefuse API Key（Litefuse Cloud 或自托管）以及你的 Anthropic API Key（Anthropic Console）。

import os
 
# Get keys for your project from the project settings page: https://litefuse.cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..."
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..."
os.environ["LANGFUSE_BASE_URL"] = "https://litefuse.cloud"
 
# Your Anthropic API key
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..."
 
# Configure LangChain OpenTelemetry instrumentation for Claude Agent SDK
os.environ["LANGSMITH_OTEL_ENABLED"] = "true"
os.environ["LANGSMITH_OTEL_ONLY"] = "true"
os.environ["LANGSMITH_TRACING"] = "true"

设置好环境变量后，我们就可以初始化 Langfuse 客户端了。get_client() 会使用环境变量中提供的凭证来初始化 Langfuse 客户端。

from langfuse import get_client
 
langfuse = get_client()
 
# Verify connection
if langfuse.auth_check():
    print("Langfuse client is authenticated and ready!")
else:
    print("Authentication failed. Please check your credentials and host.")

第 3 步：OpenTelemetry Instrumentation

使用 LangChain 的 Claude Agent SDK instrumentation 库对 Agent SDK 进行 instrument，并将 OpenTelemetry span 发送到 Litefuse。

from langsmith.integrations.claude_agent_sdk import configure_claude_agent_sdk
 
# Configure Claude Agent SDK with OpenTelemetry tracing
configure_claude_agent_sdk()

第 4 步：构建一个 Hello World Agent

每一次工具调用和模型 completion 都会被作为 OpenTelemetry span 捕获并转发到 Litefuse。

import asyncio
from claude_agent_sdk import (
    ClaudeAgentOptions,
    ClaudeSDKClient,
    tool,
    create_sdk_mcp_server,
)
from typing import Any
 
@tool(
    "get_weather",
    "Gets the current weather for a given city",
    {
        "city": str,
    },
)
async def get_weather(args: dict[str, Any]) -> dict[str, Any]:
    """Simulated weather lookup tool"""
    city = args["city"]
 
    # Simulated weather data
    weather_data = {
        "Berlin": "Cloudy, 59°F",
        "New York": "Sunny, 75°F",
    }
 
    weather = weather_data.get(city, "Weather data not available")
    return {"content": [{"type": "text", "text": f"Weather in {city}: {weather}"}]}
 
 
async def main():
    # Create SDK MCP server with the weather tool
    weather_server = create_sdk_mcp_server(
        name="weather",
        version="1.0.0",
        tools=[get_weather],
    )
 
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-5-20250929",
        system_prompt="You are a friendly travel assistant who helps with weather information.",
        mcp_servers={"weather": weather_server},
        allowed_tools=["mcp__weather__get_weather"],
    )
 
    async with ClaudeSDKClient(options=options) as client:
        await client.query("What's the weather like in Berlin and New York?")
 
        async for message in client.receive_response():
            print(message)
 
 
await main()

第 5 步：在 Litefuse 中查看 Trace

进入你的 Litefuse 控制台 → Traces。你应该能看到包含所有工具调用以及模型输入/输出的 trace。

Claude Agent SDK 在 Litefuse 中的示例 trace

Litefuse 中的 trace 链接

Interoperability with the Python SDK

You can use this integration together with the Litefuse SDKs to add additional attributes to the observation.

The @observe() decorator provides a convenient way to automatically wrap your instrumented code and add additional attributes to the observation.

from langfuse import observe, propagate_attributes, get_client
 
langfuse = get_client()
 
@observe()
def my_llm_pipeline(input):
    # Add additional attributes (user_id, session_id, metadata, version, tags) to all spans created within this execution scope
    with propagate_attributes(
        user_id="user_123",
        session_id="session_abc",
        tags=["agent", "my-observation"],
        metadata={"email": "user@litefuse.ai"},
        version="1.0.0"
    ):
 
        # YOUR APPLICATION CODE HERE
        result = call_llm(input)
 
        return result
 
# Run the function
my_llm_pipeline("Hi")

Learn more about using the Decorator in the Langfuse SDK instrumentation docs.

Troubleshooting

No observations appearing

First, enable debug mode in the Python SDK:

export LANGFUSE_DEBUG="True"

Then run your application and check the debug logs:

OTel observations appear in the logs: Your application is instrumented correctly but observations are not reaching Litefuse. To resolve this:
1. Call langfuse.flush() at the end of your application to ensure all observations are exported.
2. Verify that you are using the correct API keys and base URL.
No OTel spans in the logs: Your application is not instrumented correctly. Make sure the instrumentation runs before your application code.

Unwanted observations in Litefuse

The Langfuse SDK is based on OpenTelemetry. Other libraries in your application may emit OTel spans that are not relevant to you. These still count toward your billable units, so you should filter them out. See Unwanted spans in Litefuse for details.

Missing attributes

Some attributes may be stored in the metadata object of the observation rather than being mapped to the Litefuse data model. If a mapping or integration does not work as expected, please raise an issue on GitHub.

Next Steps

Once you have instrumented your code, you can manage, evaluate and debug your application:

Manage prompts in Litefuse Add evaluation scores Run LLM-as-a-judge Evaluators Create datasets Create custom dashboards Test queries in the Playground

BeeAI Claude Agent SDK (JS/TS)

这个页面对你有帮助吗？

支持