一、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 tracebackfrom datetime import datetime, timedeltafrom claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions
from claude_code_sdk.types import (ResultMessage, AssistantMessage, TextBlock,ToolUseBlock, ToolResultBlock
)
from claude_code_sdk._errors import CLIConnectionErroros.environ["ANTHROPIC_API_KEY"] = "你的api key"
os.environ["ANTHROPIC_BASE_URL"] = "https://api.moonshot.cn/anthropic"async def chat():"""Claude Code 聊天助手(每次请求独立客户端,避免流冲突)"""client = Noneresponses = []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("流读取冲突:可能客户端被复用或并发调用")raiseexcept 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 responsesif __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)