使用说明

LangChain MCP

LangChain Elixir的模型上下文协议（MCP）集成。该库使LangChain应用程序能够使用MCP服务器作为工具提供者，从而即时访问不断增长的MCP服务器生态系统。

📚 全部文件 -完整的API参考资料和指南可在HexDocs上获得

特性

🔌 轻松集成 -用最少的代码将MCP工具添加到您的LangChain工作流中
🛠️ 工具发现 -自动发现MCP工具并将其转换为LangChain函数
🔄 后备支持 -配置回退MCP客户端以执行弹性工具
📦 多模态内容 -完全支持文本、图像和其他内容类型
⚙️ 可配置的 -缓存工具发现、配置超时、异步执行
🧪 测试 -对单元测试的模拟支持，使用Docker的实时测试基础设施
📊 状态监测 -使用LiveView支持实时监控MCP客户端健康状况

安装

添加 langchain_mcp 您的依赖关系 mix.exs:

def deps do
  [
    {:langchain, "~> 0.4"},
    {:langchain_mcp, "~> 0.1"}
  ]
end

快速开始

1.定义MCP客户端

defmodule MyApp.GitHubMCP do
  use LangChain.MCP.Client,
    name: "MyApp",
    version: "1.0.0",
    protocol_version: "2025-03-26"
end

2.在您的监督树中启动客户端

defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      {MyApp.GitHubMCP,
       transport: {:streamable_http, base_url: "http://localhost:5000"}}
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

3.在链中使用MCP工具

alias LangChain.Chains.LLMChain
alias LangChain.ChatModels.ChatAnthropic
alias LangChain.Message
alias LangChain.MCP.Adapter

# Create adapter and discover tools
adapter = Adapter.new(client: MyApp.GitHubMCP)
mcp_functions = Adapter.to_functions(adapter)

# Mix with regular functions
all_functions = [
  MyApp.Functions.custom_function()
] ++ mcp_functions

# Use in chain
{:ok, updated_chain} =
  LLMChain.new!(%{llm: ChatAnthropic.new!()})
  |> LLMChain.add_tools(all_functions)
  |> LLMChain.add_message(Message.new_user!("Create a GitHub issue for this bug"))
  |> LLMChain.run(mode: :while_needs_response)

配置选项

适配器选项

adapter = Adapter.new(
  client: MyApp.MCPClient,
  # Cache tool discovery (default: true)
  cache_tools: true,
  # Timeout for tool calls in ms (default: 30_000)
  timeout: 30_000,
  # Mark tools as async (default: false)
  async: false,
  # Fallback client if primary fails
  fallback_client: MyApp.BackupMCPClient,
  # Filter which tools to expose (default: all)
  tool_filter: fn tool -> tool["name"] not in ["dangerous_tool"] end
)

客户类型

这 client 和 fallback_client 选项接受多种引用类型：

# Module name (most common - for supervision tree clients)
adapter = Adapter.new(client: MyApp.MCPClient)

# PID (for dynamically started clients)
{:ok, client_pid} = MyApp.MCPClient.start_link(transport: {:streamable_http, base_url: url})
adapter = Adapter.new(client: client_pid)

# Via tuple (for Registry-based lookups)
adapter = Adapter.new(client: {:via, Registry, {MyApp.Registry, "mcp_client"}})

# Global tuple
adapter = Adapter.new(client: {:global, :my_mcp_client})

动态客户

对于需要每个请求或每个作业MCP客户端的场景（例如，使用Playwright进行浏览器自动化），您可以动态启动客户端并将PID传递给适配器。

每种工作模式

defmodule MyApp.BrowserJob do
  alias LangChain.MCP.Adapter

  def run_with_browser(task) do
    # Start a dedicated MCP client for this job
    {:ok, client_pid} = MyApp.PlaywrightMCP.start_link(
      transport: {:streamable_http, base_url: "http://localhost:3000"}
    )

    # Wait for the client to be ready
    :ok = Adapter.wait_for_server_ready(client_pid)

    try do
      # Create adapter with the dynamic client
      adapter = Adapter.new(client: client_pid)
      functions = Adapter.to_functions(adapter)

      # Use in your chain
      {:ok, result} = run_chain_with_tools(task, functions)
      result
    after
      # Clean up when done
      Supervisor.stop(client_pid)
    end
  end
end

与回退

动态客户端也支持回退：

{:ok, primary_pid} = MyApp.PrimaryMCP.start_link(opts)
{:ok, fallback_pid} = MyApp.BackupMCP.start_link(opts)

adapter = Adapter.new(
  client: primary_pid,
  fallback_client: fallback_pid
)

选择性工具发现

# Get only specific tools
mcp_functions = Adapter.to_functions(adapter, only: ["search", "fetch"])

# Exclude certain tools
mcp_functions = Adapter.to_functions(adapter, except: ["admin_tool"])

状态监测

实时监控MCP客户端的健康状况和状态。

注册客户

从应用程序模块注册客户端：

defmodule MyApp.Application do
  use Application
  alias LangChain.MCP.StatusMonitor

  def start(_type, _args) do
    children = [
      {MyApp.GitHubMCP, transport: {:streamable_http, base_url: "http://localhost:5000"}}
    ]

    {:ok, _pid} = Supervisor.start_link(children, strategy: :one_for_one)

    # Register clients for status monitoring
    StatusMonitor.register_client_by_name(MyApp.GitHubMCP, :github)

    {:ok, self()}
  end
end

查询客户端状态

alias LangChain.MCP.StatusMonitor

# Check if registered
StatusMonitor.registered?(:github)
# => true

# Get client status
StatusMonitor.get_client_status(:github)
# => {:ok, %{pid: #PID<0.123.0>}}

# Get health summary
StatusMonitor.health_summary()
# => %{
#   healthy_clients: [:github, :filesystem],
#   unhealthy_clients: [],
#   total_clients: 2,
#   uptime_percentage: 100.0
# }

# Get dashboard-ready status
StatusMonitor.dashboard_status()

Phoenix LiveView集成

完整的LiveView示例用于实时状态监控：

自动刷新仪表板（每2秒一次）
颜色编码的健康指标
正常运行时间百分比跟踪
客户详细信息

看 examples/phoenix_liveview_example.ex 和 STATUS_MONITOR.md 以获取完整的文档。

后备支持

与LangChain的LLM回退类似，您可以配置回退MCP客户端：

adapter = Adapter.new(
  client: MyApp.PrimaryMCP,
  fallback_client: MyApp.BackupMCP,
  # Optional: modify behavior before fallback
  before_fallback: fn _adapter, tool_name, _args ->
    Logger.warning("Falling back for tool: #{tool_name}")
    :continue  # or :skip to skip fallback
  end
)

测试

使用模拟进行单元测试

defmodule MyApp.MyAgentTest do
  use ExUnit.Case

  setup do
    # Use Anubis mock transport for testing
    {:ok, client} = MyApp.TestMCP.start_link(
      transport: {:mock, responses: %{
        "list_tools" => %{"tools" => [...]},
        "call_tool" => %{"content" => [...]}
      }}
    )

    {:ok, client: client}
  end

  test "agent uses MCP tools", %{client: _client} do
    # Your test here
  end
end

实时集成测试

这个项目包括一个内置的基于Elixir的MCP测试服务器——不需要Docker！

步骤1：启动测试服务器

mix test_server

这将在上启动MCP服务器 http://localhost:5000 使用测试工具：

get_current_time -获取UTC或指定时区的当前时间
get_timestamp -获取当前Unix时间戳
add_numbers -将两个数字相加

步骤2：运行集成测试 （在单独的终端中）

mix test --include live_call

自定义端口： 如果您需要使用其他端口：

# Terminal 1: Start server on custom port
mix test_server --port 5000

# Terminal 2: Run tests with custom URL
MCP_TEST_URL=http://localhost:5000 mix test --include live_call

现场测试示例：

@tag :live_call
test "integrates with real MCP server" do
  adapter = Adapter.new(client: MyApp.MCPClient)
  functions = Adapter.to_functions(adapter)

  assert length(functions) > 0
end

使用Docker（可选）

如果你更喜欢Docker或需要在外部MCP服务器上进行测试，你可以使用Docker compose：

# Start Docker-based MCP servers
docker-compose up -d

# Run tests
mix test --include live_call

注：大多数官方MCP服务器（如 mcp_server_time)使用stdio传输，没有包装器就无法通过HTTP访问。建议使用内置的Elixir测试服务器。

可用的MCP服务器

许多MCP服务器都可以通过Docker Hub获得：

mcp/time -时间和时区操作
mcp/github -GitHub API集成
mcp/postgres -PostgreSQL数据库访问
mcp/sqlite -SQLite数据库操作
mcp/puppeteer -浏览器自动化

看 Docker Hub MCP目录更多。

贡献

质量检查

在推送之前，确保代码符合质量标准：

# Run all checks (same as CI)
mix quality_check

# Or run individually:
mix format --check-formatted    # Code formatting  
mix credo --strict              # L linting
mix dialyzer                    # Type checking
mix test --exclude live_call    # Unit tests only

自动发布工作流要求通过这些检查。使用 mix quality_check 在当地承诺之前。

测试

单元测试：带模拟通孔的标准ExUnit :mimic
集成测试：标记 :live_call，需要运行测试服务器

# Start test server (Terminal 1)
mix test_server

# Run integration tests (Terminal 2)  
mix test --include live_call

建筑

┌─────────────────────────┐
│   LangChain.LLMChain    │
│   + tools: [Function]   │
└────────────┬────────────┘
             │
             ├─── Regular Functions
             │
             └─── MCP Functions (via Adapter)
                  │
                  ▼
         ┌────────────────────┐
         │  LangChain.MCP     │
         │  • Adapter         │
         │  • SchemaConverter │
         │  • ToolExecutor    │
         │  • ContentMapper   │
         │  • ErrorHandler    │
         └────────┬───────────┘
                  │
                  ▼
         ┌────────────────────┐
         │  Anubis.Client     │
         │  (GenServer)       │
         └────────┬───────────┘
                  │
                  ▼
         ┌────────────────────┐
         │   MCP Server       │
         │ (Docker/External)  │
         └────────────────────┘

许可证

Apache 2.0许可证。有关详细信息，请参阅许可证。

LangChain MCP集成工具

使用说明

LangChain MCP

特性

安装

快速开始

1.定义MCP客户端

2.在您的监督树中启动客户端

3.在链中使用MCP工具

配置选项

适配器选项

客户类型

动态客户

每种工作模式

与回退

选择性工具发现

状态监测

注册客户

查询客户端状态

Phoenix LiveView集成

后备支持

测试

使用模拟进行单元测试

实时集成测试

使用Docker（可选）

可用的MCP服务器

贡献

质量检查

测试

建筑

许可证

链接

使用说明

LangChain MCP

特性

安装

快速开始

1.定义MCP客户端

2.在您的监督树中启动客户端

3.在链中使用MCP工具

配置选项

适配器选项

客户类型

动态客户

每种工作模式

与回退

选择性工具发现

状态监测

注册客户

查询客户端状态

Phoenix LiveView集成

后备支持

测试

使用模拟进行单元测试

实时集成测试

使用Docker（可选）

可用的MCP服务器

贡献

质量检查

测试

建筑

许可证

链接

相关 MCP

ChatGPT交互工具

Loki日志查询服务

Genkit MCP 集成插件

Git MCP 服务器

WHOIS查询服务

AI助手协议服务器