一、MCP配置前期准备
（一）创建个人令牌/群组令牌
我这里是创建个人令牌，去到首页左上角，点击头像——>偏好设置——>访问令牌——>添加新令牌
（二）配置mcp信息
去到魔塔社区，点击mcp广场，然后搜索gitlab，把刚刚生成的个人令牌粘贴进去
这里的url如果是你自己部署的话，就替换前面的域名即可，比如https://gitlab.com/api/v4就换成http://ip:端口/api/v4，配置完后就会生成对应的sse配置信息
二、Claude Code SDK 配置
这里有一个巨巨巨巨巨坑，正常我们在终端使用claude的时候，偶尔会弹出让你是否确认创建某个文件夹或者其他的操作等信息，如下：
这是cc的一个权限机制，在claude code的文档里面也有提到：
更详细的可以看这篇文章：Claude Code权限模式详解：Default、AcceptEdits、Plan、BypassPermissions四种模式 - 博客 - Hrefgo AI
（一）代码示例
import asyncio
import os
import traceback
from datetime import datetime, timedelta
from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions
from claude_code_sdk.types import (
ResultMessage, AssistantMessage, TextBlock,
ToolUseBlock, ToolResultBlock
)
from claude_code_sdk._errors import CLIConnectionError
os.environ["ANTHROPIC_API_KEY"] = "你的api key"
os.environ["ANTHROPIC_BASE_URL"] = "https://api.moonshot.cn/anthropic"
async def chat():
"""
Claude Code 聊天助手（每次请求独立客户端，避免流冲突）
"""
client = None
responses = []
try:
# 每次请求都创建新客户端（避免复用导致的流冲突）
mcp_servers = {
"mcp-gitlab-server": {
"type": "sse",
"url": "你在魔塔生成的url"
}
}
options = ClaudeCodeOptions(
cwd=".",
permission_mode="bypassPermissions", # 绕过权限(!很重要，不然执行不了)
mcp_servers=mcp_servers
)
client = ClaudeSDKClient(options=options)
# 连接
await client.connect()
prompt = "使用mcp-gitlab-server这个mcp工具帮我在gitlab仓库中创建一个名为camel_test的项目"
await client.query(prompt, session_id="123456")
try:
async for message in client.receive_messages():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
responses.append({
"role": "assistant",
"content": block.text.strip(),
"type": "text"
})
print({
"role": "assistant",
"content": block.text.strip(),
"type": "text"
})
elif isinstance(block, ToolUseBlock):
responses.append({
"role": "assistant",
"content": f"使用工具: {block.name}",
"type": "tool",
"metadata": {"tool_name": block.name, "parameters": block.input}
})
print({
"role": "assistant",
"content": f"使用工具: {block.name}",
"type": "tool",
"metadata": {"tool_name": block.name, "parameters": block.input}
})
elif isinstance(message, ToolResultBlock):
status = "成功" if not message.is_error else "失败"
responses.append({
"role": "system",
"content": f"工具执行{status}: {message.content}",
"type": "tool_result",
"metadata": {"is_error": message.is_error, "tool_use_id": message.tool_use_id}
})
print({
"role": "system",
"content": f"工具执行{status}: {message.content}",
"type": "tool_result",
"metadata": {"is_error": message.is_error, "tool_use_id": message.tool_use_id}
})
elif isinstance(message, ResultMessage):
responses.append({
"role": "system",
"content": "本轮响应结束",
"type": "result",
"metadata": {
"input_tokens": message.usage.get("input_tokens"),
"output_tokens": message.usage.get("output_tokens"),
"cost_usd": message.total_cost_usd,
"duration_ms": message.duration_ms
}
})
print({
"role": "system",
"content": "本轮响应结束",
"type": "result",
"metadata": {
"input_tokens": message.usage.get("input_tokens"),
"output_tokens": message.usage.get("output_tokens"),
"cost_usd": message.total_cost_usd,
"duration_ms": message.duration_ms
}
})
break # 结束接收
except Exception as e:
if "another coroutine is already waiting" in str(e):
print("流读取冲突：可能客户端被复用或并发调用")
raise
except CLIConnectionError:
raise Exception("无法连接到 Claude 服务，请检查网络或 API 密钥配置")
except Exception as e:
print(f"聊天请求失败: {e}")
traceback.print_exc()
raise Exception(f"内部错误: {str(e)}")
finally:
# 确保关闭客户端
if client:
try:
await client.disconnect()
except:
pass # 忽略关闭时的异常
return responses
if __name__ == "__main__":
# 生产环境建议使用 gunicorn + uvicorn 部署
start_time = datetime.now()
asyncio.run(chat())
print(f"总耗时: {(datetime.now() - start_time).total_seconds()} 秒")
（二）重要配置
options = ClaudeCodeOptions(
cwd=".",
permission_mode="bypassPermissions", # 绕过权限(!很重要，不然执行不了)
mcp_servers=mcp_servers
)（三）运行结果
