此MCP(模型上下文协议)服务器提供了Merge API与任何支持MCP协议的LLM提供商之间的集成(例如,桌面版Claude),使您能够使用自然语言与您的Merge数据进行交互。
使用独立安装程序安装uv:
# 在macOS和Linux上。
curl -LsSf https://astral.sh/uv/install.sh | sh
# 在Windows上。
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
或者通过pip安装:
# 使用pip。
pip install uv
# 使用pipx。
pipx install uv
这是一个示例配置文件,您可以使用它来设置Merge MCP。
{
"mcpServers": {
"merge-mcp-server": {
"command": "uvx",
"args": ["merge-mcp"],
"env": {
"MERGE_API_KEY": "your_api_key",
"MERGE_ACCOUNT_TOKEN": "your_account_token"
}
}
}
}
注意:如果“uvx”命令不起作用,请尝试绝对路径(例如 /Users/username/.local/bin/uvx)
uvxclaude_desktop_config.json的文件。your_api_key和your_account_token替换为您实际的Merge API密钥和链接账户令牌。您还需要在配置文件中用绝对路径替换uvx(例如/Users/username/.local/bin/uvx)。您可以通过运行终端中的which uvx找到绝对路径。# 创建项目目录
mkdir mcp-client
cd mcp-client
# 创建虚拟环境
python -m venv .venv
# 激活虚拟环境
# 在Windows上:
.venv\Scripts\activate
# 在Unix或MacOS上:
source .venv/bin/activate
# 安装所需包
pip install mcp uv anthropic python-dotenv
# 创建我们的主文件
touch client.py
# 将您的ANTHROPIC_API_KEY和MERGE_API_KEY添加到.env
echo "ANTHROPIC_API_KEY=<您的Anthropic密钥>" >> .env
echo "MERGE_API_KEY=<您的Merge密钥>" >> .env
# 将.env文件添加到.gitignore
echo ".env" >> .gitignore
client.py文件并添加以下代码import os
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv() # 从.env加载环境变量
class MCPClient:
def __init__(self):
# 初始化会话和客户端对象
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.anthropic = Anthropic()
# 方法将放在这里
connect_to_server函数async def connect_to_server(self, linked_account_token: str):
"""连接到MCP服务器
参数:
linked_account_token: 相关链接账户的令牌
"""
server_params = StdioServerParameters(
command="uvx",
args=["merge-mcp"],
env={
"MERGE_API_KEY": os.getenv("MERGE_API_KEY"),
"MERGE_ACCOUNT_TOKEN": linked_account_token
}
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# 列出可用工具
response = await self.session.list_tools()
tools = response.tools
print("\n已连接到具有以下工具的服务器:", [tool.name for tool in tools])
process_query函数async def process_query(self, query: str) -> str:
"""使用Claude和可用工具处理查询"""
messages = [
{
"role": "user",
"content": query
}
]
response = await self.session.list_tools()
available_tools = [{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
} for tool in response.tools]
# 初始Claude API调用
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=messages,
tools=available_tools
)
# 处理响应并处理工具调用
final_text = []
assistant_message_content = []
for content in response.content:
if content.type == 'text':
final_text.append(content.text)
assistant_message_content.append(content)
elif content.type == 'tool_use':
tool_name = content.name
tool_args = content.input
# 获取工具调用执行确认
confirmation = input(f"是否要调用工具'{tool_name}',参数{tool_args}?(y/n): ").strip().lower()
if confirmation.startswith('y'):
result = await self.session.call_tool(tool_name, tool_args)
final_text.append(f"[调用工具{tool_name},参数{tool_args}]")
assistant_message_content.append(content)
messages.append({
"role": "assistant",
"content": assistant_message_content
})
messages.append({
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": content.id,
"content": result.content
}
]
})
# 从Claude获取下一个响应
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=messages,
tools=available_tools
)
final_text.append(response.content[0].text)
else:
final_text.append(f"[跳过调用工具{tool_name},参数{tool_args}]")
return "\n".join(final_text)
chat_loop函数 async def chat_loop(self):
"""运行交互式聊天循环"""
print("\nMCP客户端已启动!")
print("输入您的查询或输入'quit'退出。")
while True:
try:
query = input("\n查询: ").strip()
if query.lower() == 'quit':
break
response = await self.process_query(query)
print("\n" + response)
except Exception as e:
print(f"\n错误: {str(e)}")
cleanup函数 async def cleanup(self):
"""清理资源"""
await self.exit_stack.aclose()
client.py文件中添加一个main函数作为主要入口点async def main():
client = MCPClient()
try:
await client.connect_to_server("<您的链接账户令牌>")
await client.chat_loop()
finally:
await client.cleanup()
if __name__ == "__main__":
import sys
asyncio.run(main())
python client.py
范围决定了哪些工具在MCP服务器上启用,并用于控制对Merge API不同部分的访问。如果没有指定范围,则所有可用范围都将被启用。
启动服务器时,您可以指定要启用的范围。这是通过传递带有范围列表的--scopes标志来完成的。
{
"mcpServers": {
"merge-mcp-server": {
"command": "uvx",
"args": [
"merge-mcp",
"--scopes",
"ats.Job:read",
"ats.Candidate",
"ats.Application:write"
],
"env": {
"MERGE_API_KEY": "your_api_key",
"MERGE_ACCOUNT_TOKEN": "your_account_token"
}
}
}
}
Merge MCP服务器中的范围遵循基于Merge API类别和常见模型名称的特定格式。每个范围格式如下:
<category>.<CommonModel>:<permission>
其中:
<category> 是Merge API类别(例如,hris,ats,accounting)<CommonModel> 是Merge通用模型的名称(例如,Employee,Candidate,Account)<permission> 是read或write(可选 - 如果未指定,则授予所有权限)有效范围的示例:
hris.Employee:read - 允许读取HRIS类别中的员工数据ats.Candidate:write - 允许在ATS类别中创建或更新候选人数据accounting.Account - 允许对会计类别中的账户数据进行所有操作您可以组合多个范围以授予不同的权限。
可用范围取决于您的Merge API账户配置以及链接账户可以访问的模型。范围必须与您的链接账户上的启用范围交叉引用:
ats.Job与HRIS链接账户),则不会返回该范围的任何工具。hris.Employee:write),则需要该权限的工具将不会返回。范围通常对应于Merge API中的不同模型或实体类型,并且它们控制这些实体的读取和写入访问。
Merge MCP服务器提供了对各种Merge API端点作为工具的访问。可用工具取决于您的Merge API类别(HRIS、ATS等)和您启用的范围。
工具是根据您的Merge API模式动态生成的,并包括以下操作:
注意:目前不支持下载工具。这是一个已知的限制,将在未来的版本中解决。
Merge MCP服务器使用的环境变量如下:
MERGE_API_KEY:您的Merge API密钥MERGE_ACCOUNT_TOKEN:您的Merge链接账户令牌MERGE_TENANT(可选):Merge API租户。有效值为US,EU和APAC。默认为US。