Kedro MCP Agent

凯德罗·麦克普特工

这个仓库是一个小型游乐场，它将一个标准的Kedro示例项目与一个MCP服务器和两个不同的MCP代理相结合。

基础项目结构是通过以下方式生成的 kedro new 使用Spaceflights示例的命令。如果你想全面了解Kedro的概念（项目布局、管道、配置、CLI等），请按照官方课程而不是本自述文件进行操作：

https://docs.kedro.org/en/stable/getting-started/course/

下面我们只关注MCP/MCP代理部分： mcp_server.py, agent.py，以及 agent_langgraph.py.

______________________________________________________________________

与VS代码代理模式一起使用（推荐）

使用此项目的主要方法是运行MCP服务器并将其连接到VS Code的代理模式。服务器包装您的Kedro项目，以便VS Code中的代理可以作为工具发现和执行管道——所有这些都在您的首选环境中运行。

1.启动MCP服务器

python mcp_server.py --transport streamable-http

这将在以下位置启动服务器 http://localhost:8001/mcp.

2.连接VS代码

1. 打开VS代码。
2. 按 ⇧⌘P （macOS）/ Ctrl+Shift+P （Windows/Linux）打开命令面板。
3. 选择 MCP：打开用户配置.
4. 添加以下配置：

{
  "servers": {
    "kedro": {
      "url": "http://localhost:8001/mcp"
    }
  }
}

1. 切换到 代理模式 在Copilot聊天面板中。

VS Code将连接到正在运行的服务器并公开Kedro工具(list_pipelines, get_pipeline_info, run_pipeline, list_datasets)给代理人。然后，您可以要求代理检查管道、运行管道、列出数据集等。--它将调用引擎盖下的MCP工具。

______________________________________________________________________

mcp_server.py –Kedro项目上的MCP服务器

mcp_server.py 使用以下工具将Kedro项目转换为模型上下文协议（MCP）服务器 FastMCP 类从 mcp 包裹：

• 使用a 工厂模式 (create_mcp_server)启动Kedro项目并返回配置好的 FastMCP 例子
• 包括a 共享执行核心 (_execute_pipeline)返回结构化结果(run_id, status, duration_ms)带有可选的错误详细信息和调试回溯。
• 显示四个MCP工具：

- list_pipelines –返回已注册管道名称的JSON数组。 - get_pipeline_info –返回给定管道的详细节点级信息（输入、输出、标签）。 - run_pipeline –执行具有完整参数支持的管道（标签、节点选择、从/到节点、运行器选择、命名空间、， only_missing_outputs等等）。 - list_datasets –列出Kedro数据目录中的所有数据集。

• 支持 多个传输 可通过CLI标志进行配置：

- stdio （默认）-供MCP客户端基于子流程使用。 - sse –服务器通过HTTP发送事件。 - streamable-http –流式HTTP传输。

• 调试模式 (--debug 或 KEDRO_SERVER_DEBUG=1)在错误响应中包含堆栈跟踪。

默认情况下，代理通过stdio将服务器作为子进程启动。您也可以独立运行它：

python mcp_server.py                                    # stdio (default)
python mcp_server.py --transport sse                     # SSE on 127.0.0.1:8001
python mcp_server.py --transport streamable-http --port 9000
python mcp_server.py --debug                             # include tracebacks

______________________________________________________________________

agent.py –MCP代理 使用Langfuse追踪

agent.py 是一个简单但完全可追踪的MCP代理：

• 开始 mcp_server.py 作为stdio上的子流程，打开MCP ClientSession.
• 呼叫 session.list_tools() 以发现服务器暴露的工具。
• 在OpenAI聊天完成API之上实现手动工具调用循环：

- 将MCP工具模式转换为OpenAI工具定义。 - 将用户提示和工具定义发送到模型。 - 读取 tool_calls 根据模型的响应，在MCP服务器上执行它们 session.call_tool. - 将工具输出反馈到LLM，直到它返回最终的自然语言答案。

• 合并 廊坊 以“老式”的方式实现可观测性：

- 进口 AsyncOpenAI 从 langfuse.openai. - 所有呼叫 self.llm.chat.completions.create(...) 假设正常，则会自动跟踪到您的Langfuse项目 LANGFUSE_* 和 OPENAI_API_KEY 设置环境变量。

如果你想看到MCP+Kedro+OpenAI+Langfuse的端到端集成，而不需要任何LangChain/LangGraph抽象，那么这个文件就是参考实现。

从激活的虚拟环境运行它：

python agent.py

您将收到一个交互式提示，您可以在其中提问；代理将通过MCP工具检查和运行Kedro管道，所有LLM调用都将在Langfuse中显示为痕迹。

______________________________________________________________________

agent_langgraph.py –基于LangGraph的MCP代理（无Langfuse）

agent_langgraph.py 是基于LangGraph和LangChain构建的替代代理。它保持相同的MCP服务器，但将编排委托给基于图的代理：

• 启动并连接到 mcp_server.py 通过MCP stdio客户端（与 agent.py).
• 用途 langchain_mcp_adapters.tools.load_mcp_tools 将MCP工具自动转换为LangChain工具。
• 使用LangGraph/LangChain创建ReAct样式代理：

- LLM ChatOpenAI 从 langchain_openai. - 工具：返回的MCP工具 load_mcp_tools.

• 提供：

- process_query(...) –通过LangGraph代理运行单个查询。 - chat() -带有对话记忆的交互式循环（状态将之前的消息轮流传递）。 - stream_query(...) –在生成部分响应时进行流式传输。

运行它：

python agent_langgraph.py

您将获得一个基于LangGraph的会话代理，它仍然可以通过MCP检查和运行Kedro管道。