主题
LangSmith Tool Server
LangSmith Tool Server 是一个独立的 MCP 框架,用于构建和部署带有内置身份验证和授权的工具。当你需要以下功能时,请使用 Tool Server:
- 创建自定义工具,与 LangSmith 的 Agent Auth 集成以实现 OAuth 身份验证
- 为你自己构建的 agent(Fleet 之外)构建 MCP 网关
NOTE
如果你正在使用 Fleet,则无需直接与 Tool Server 交互。Fleet 提供内置工具并支持远程 MCP 服务器,无需设置 Tool Server。
不过,你可以将关联的 tool server 实例配置为 MCP 服务器,这样就能在 agent 中使用自定义 MCP 服务器。
下载 PyPI 包以开始使用。
创建自定义工具包
安装 LangSmith Tool Server 和 LangChain CLI:
bash
pip install langsmith-tool-server
pip install langchain-cli-v2创建新的工具包:
bash
langchain tools new my-toolkit
cd my-toolkit这会创建一个具有以下结构的工具包:
my-toolkit/
├── pyproject.toml
├── toolkit.toml
└── my_toolkit/
├── __init__.py
├── auth.py
└── tools/
├── __init__.py
└── ...使用 @tool 装饰器定义工具。有关工具 schema、返回值、错误处理和 ToolRuntime 的更多信息,请参阅工具指南。
python
from langsmith_tool_server import tool
@tool
def hello(name: str) -> str:
"""Greet someone by name."""
return f"Hello, {name}!"
@tool
def add(x: int, y: int) -> int:
"""Add two numbers."""
return x + y
TOOLS = [hello, add]运行服务器:
bash
langchain tools serve你的 tool server 将在 http://localhost:8000 上启动。
通过 MCP 协议调用工具
以下示例列出可用工具并调用 add 工具:
python
import asyncio
import aiohttp
async def mcp_request(url: str, method: str, params: dict = None):
async with aiohttp.ClientSession() as session:
payload = {"jsonrpc": "2.0", "method": method, "params": params or {}, "id": 1}
async with session.post(f"{url}/mcp", json=payload) as response:
return await response.json()
async def main():
url = "http://localhost:8000"
tools = await mcp_request(url, "tools/list")
print(f"Tools: {tools}")
result = await mcp_request(url, "tools/call", {"name": "add", "arguments": {"a": 5, "b": 3}})
print(f"Result: {result}")
asyncio.run(main())用作 MCP 网关
LangSmith Tool Server 可以充当 MCP 网关,将多个 MCP 服务器中的工具聚合到单个端点。在 toolkit.toml 中配置 MCP 服务器:
toml
[toolkit]
name = "my-toolkit"
tools = "./my_toolkit/__init__.py:TOOLS"
[[mcp_servers]]
name = "weather"
transport = "streamable_http"
url = "http://localhost:8001/mcp/"
[[mcp_servers]]
name = "math"
transport = "stdio"
command = "python"
args = ["-m", "mcp_server_math"]来自已连接 MCP 服务器的所有工具都会通过你服务器的 /mcp 端点暴露。MCP 工具会以其服务器名称为前缀以避免冲突(例如 weather_get_forecast、math_add)。
身份认证
第三方 API 的 OAuth
对于需要访问第三方 API 的工具(如 Google、GitHub、Slack 等),你可以使用 Agent Auth 进行 OAuth 身份验证。
在工具中使用 OAuth 之前,你需要在 LangSmith 工作区设置中配置 OAuth 提供商。设置说明请参阅 Agent Auth 文档。
配置完成后,在工具装饰器中指定 auth_provider:
python
from langsmith_tool_server import tool, Context
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
@tool(
auth_provider="google",
scopes=["https://www.googleapis.com/auth/gmail.readonly"],
integration="gmail"
)
async def read_emails(context: Context, max_results: int = 10) -> str:
"""Read recent emails from Gmail."""
credentials = Credentials(token=context.token)
service = build('gmail', 'v1', credentials=credentials)
# ... Gmail API calls
return f"Retrieved {max_results} emails"带有 auth_provider 的工具必须:
- 将
context: Context作为第一个参数 - 指定至少一个 scope
- 使用
context.token发起经过身份验证的 API 调用
自定义请求身份验证
自定义身份验证允许你验证请求并与身份提供商集成。在 auth.py 文件中定义身份验证处理程序:
python
from langsmith_tool_server import Auth
auth = Auth()
@auth.authenticate
async def authenticate(authorization: str = None) -> dict:
"""Validate requests and return user identity."""
if not authorization or not authorization.startswith("Bearer "):
raise auth.exceptions.HTTPException(
status_code=401,
detail="Unauthorized"
)
token = authorization.replace("Bearer ", "")
# Validate token with your identity provider
user = await verify_token_with_idp(token)
return {"identity": user.id}该处理程序会在每个请求上运行,并且必须返回一个包含 identity(以及可选的 permissions)的 dict。