基于MCP协议实现端到端全链路追踪

1. 前置条件

Python版本 ≥ 3.8

2. 准备环境变量

为 Python 应用配置环境变量，另外演示场景中需要调用大模型处理用户问题，此处以百炼大模型为例，环境变量中还需配置百炼API-Key，

export ARMS_APP_NAME=xxx   # 应用名称。
export ARMS_REGION_ID=xxx   # 对应的阿里云账号的RegionID。
export ARMS_LICENSE_KEY=xxx   # 阿里云 LicenseKey。
export ARMS_IS_PUBLIC=True   # 添加该环境变量表示使用公网访问。
export BALIAN_API_KEY=sk-************   # 百炼API-Key。

您可以通过DescribeTraceLicenseKey OpenAPI获取阿里云LicenseKey。

3. 安装 Python 依赖

# 安装项目依赖
pip install mcp==1.10.1 langchain-mcp-adapters langgraph "langchain[openai]" aliyun-bootstrap rich mcp_server_aliyun_observability -y

# 安装 ARMS Python 探针
aliyun-bootstrap -a install

1. 启动 SLS MCP 服务器

在终端中执行以下命令，在本地启动 SLS MCP 服务器。获取 AccessKey 的方法，请参见创建AccessKey。

aliyun-instrument python -m mcp_server_aliyun_observability \
  --transport sse \
  --access-key-id <阿里云AccessKey ID> \
  --access-key-secret <阿里云AccessKey Secret>

当看到如下日志输出时，说明 MCP 服务器已启动，ARMS Python 探针已经成功注入到 MCP 服务器中。

Load Aliyun python agent version: 1.6.1

2. 启动 Langchain Demo 程序

创建 demo.py 文件，执行 aliyun-instrument python demo.py 来启动此 Demo 程序。

此处示例Demo创建了一个 Agent，并向 Agent 提了两个问题，该 Agent 会使用 SLS MCP 服务处理日志分析请求。

from contextlib import asynccontextmanager
from mcp import ClientSession
from mcp.client.sse import sse_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
import asyncio
from opentelemetry import trace
import os

tracer = trace.get_tracer(__name__)

# 创建一个大模型对象
model = ChatOpenAI(
    model="qwen-turbo",
    api_key=os.getenv("BALIAN_API_KEY"),  # 百炼 API Key
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 创建 agent
@asynccontextmanager
async def sls_agent():
    with tracer.start_as_current_span("langchain-demo"):
        async with sse_client(url="http://0.0.0.0:8000/sse") as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                tools = await load_mcp_tools(session)
                agent = create_react_agent(model, tools)
                yield agent

def print_response(response):
    for message in response['messages']:
        print(message.content)

async def main():
    # 创建一个 sls agent
    async with sls_agent() as agent:
        # 调用模型处理问题
        response = await agent.ainvoke(
            {"messages": "我要查询我在杭州地域的project的列表，取前10个"}
        )
        print_response(response)

        # 继续追问
        response = await agent.ainvoke(
            {
                "messages": "查询杭州地域下 demo-test 这个 project 下的 loghub_master 日志库的日志，分析最近十分钟写入了多少条日志"
            }
        )
        print_response(response)

if __name__ == "__main__":
    asyncio.run(main())

程序将会输出如下结果：

我要查询我在杭州地域的project的列表，取前10个

以下是您在杭州地域（cn-hangzhou）的前10个项目列表：

1. sls-aysls-pub-cn-hangzhou-b-xxxx
2. e2e-test-xxxx
3. workspace-xxxx
4. ws-xxxx (描述: Create by o11y)
5. clickbench8 (描述: clickbench8)
6. clickbench16 (描述: clickbench16)
7. clickbench (描述: clickbench)
8. job-e2e-project-xxxx
9. job-e2e-project-xxxx
10. job-e2e-project-xxxx
如果需要查询更多项目，您可以提供项目的关键词进行模糊搜索。

查询杭州地域下 demo-test 这个 project 下的 loghub_master 日志库的日志，分析最近十分钟写入了多少条日志

在最近的十分钟内，`demo-test` 项目下的 `loghub_master` 日志库共写入了 **8641 条日志**。