你对Claude说话
  → Claude决定要调用某个外部工具
  → Claude Code（Client）通过MCP协议向Server发送工具调用请求
  → MCP Server收到请求，调用真实的外部API（GitHub/数据库/Slack等）
  → Server把结果返回给Claude Code
  → Claude看到结果，继续和你对话

# 添加GitHub MCP（通过npx运行）
claude mcp add github -- npx @modelcontextprotocol/server-github

# 添加MySQL MCP
claude mcp add mysql -- npx @modelcontextprotocol/server-mysql

# 添加HTTP类型的MCP
claude mcp add --transport http myserver https://mcp.example.com

# 查看已添加的MCP列表
claude mcp list

# 删除某个MCP
claude mcp remove github

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
      }
    },
    "mysql": {
      "command": "npx",
      "args": ["-y", "@benborla29/mcp-server-mysql"],
      "env": {
        "MYSQL_HOST": "localhost",
        "MYSQL_PORT": "3306",
        "MYSQL_USER": "readonly_user",
        "MYSQL_PASSWORD": "${DB_PASSWORD}",
        "MYSQL_DATABASE": "myapp_prod",
        "MYSQL_ALLOW_INSERT_OPERATION": "false",
        "MYSQL_ALLOW_UPDATE_OPERATION": "false",
        "MYSQL_ALLOW_DELETE_OPERATION": "false"
      }
    },
    "sentry": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-sentry"],
      "env": {
        "SENTRY_AUTH_TOKEN": "${SENTRY_TOKEN}",
        "SENTRY_ORG": "your-org-slug"
      }
    }
  }
}

{
  "mcpServers": {
    "postgresql": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "POSTGRES_CONNECTION_STRING": "postgresql://readonly_user:${PG_PASSWORD}@localhost:5432/myapp_prod"
      }
    }
  }
}

{
  "mcpServers": {
    "redis": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-redis"],
      "env": {
        "REDIS_URL": "redis://:${REDIS_PASSWORD}@localhost:6379/0"
      }
    }
  }
}

{
  "mcpServers": {
    "jira": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-jira"],
      "env": {
        "JIRA_URL": "https://your-company.atlassian.net",
        "JIRA_EMAIL": "your-email@company.com",
        "JIRA_API_TOKEN": "${JIRA_TOKEN}"
      }
    }
  }
}

{
  "mcpServers": {
    "slack": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-slack"],
      "env": {
        "SLACK_BOT_TOKEN": "${SLACK_BOT_TOKEN}",
        "SLACK_TEAM_ID": "T01234567"
      }
    }
  }
}

{
  "mcpServers": {
    "docs-library": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/you/company-docs",
        "/Users/you/api-specs"
      ]
    }
  }
}

pip install mcp httpx

#!/usr/bin/env python3
"""
公司内部部署平台的MCP Server
暴露两个工具：
- get_deploy_status：查看某个服务的部署状态
- trigger_deploy：触发一次部署
"""
import asyncio
import json
from typing import Any
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

# 公司内部部署平台的API地址（从环境变量读取，不硬编码）
import os
DEPLOY_API_BASE = os.environ.get("DEPLOY_API_BASE", "https://deploy.internal.company.com")
DEPLOY_API_TOKEN = os.environ.get("DEPLOY_API_TOKEN", "")

# 创建MCP Server实例
server = Server("deploy-platform")

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """向Claude注册这个Server提供的工具列表"""
    return [
        types.Tool(
            name="get_deploy_status",
            description=(
                "查询指定服务的最新部署状态，包括部署时间、部署人、版本号、当前状态（成功/失败/进行中）。"
                "当用户问某个服务的部署情况、上线状态、当前版本时调用。"
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "service_name": {
                        "type": "string",
                        "description": "服务名称，例如 user-service、order-service"
                    },
                    "environment": {
                        "type": "string",
                        "enum": ["dev", "staging", "prod"],
                        "description": "环境，默认是prod",
                        "default": "prod"
                    }
                },
                "required": ["service_name"]
            }
        ),
        types.Tool(
            name="trigger_deploy",
            description=(
                "触发指定服务的部署。会用指定的镜像版本或者Git Tag发起一次部署。"
                "注意：只能部署到dev和staging环境，prod环境需要在部署平台手动确认。"
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "service_name": {
                        "type": "string",
                        "description": "要部署的服务名称"
                    },
                    "version": {
                        "type": "string",
                        "description": "镜像Tag或者Git Tag，例如 v1.2.3 或 latest"
                    },
                    "environment": {
                        "type": "string",
                        "enum": ["dev", "staging"],
                        "description": "目标环境（只支持dev和staging）"
                    },
                    "reason": {
                        "type": "string",
                        "description": "部署原因，会记录在部署日志里"
                    }
                },
                "required": ["service_name", "version", "environment"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
    """处理Claude发来的工具调用请求"""

    if name == "get_deploy_status":
        return await _get_deploy_status(
            service_name=arguments["service_name"],
            environment=arguments.get("environment", "prod")
        )

    elif name == "trigger_deploy":
        return await _trigger_deploy(
            service_name=arguments["service_name"],
            version=arguments["version"],
            environment=arguments["environment"],
            reason=arguments.get("reason", "Claude触发的部署")
        )

    else:
        return [types.TextContent(
            type="text",
            text=f"未知工具：{name}"
        )]

async def _get_deploy_status(service_name: str, environment: str) -> list[types.TextContent]:
    """调用部署平台API查询状态"""
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                f"{DEPLOY_API_BASE}/api/v1/deployments/{service_name}",
                params={"env": environment},
                headers={"Authorization": f"Bearer {DEPLOY_API_TOKEN}"},
                timeout=10.0
            )
            response.raise_for_status()
            data = response.json()

            # 格式化成易读的文本返回给Claude
            result = (
                f"服务：{service_name}（{environment} 环境）\n"
                f"状态：{data['status']}\n"
                f"版本：{data['version']}\n"
                f"部署时间：{data['deployed_at']}\n"
                f"部署人：{data['deployed_by']}\n"
                f"提交信息：{data.get('commit_message', 'N/A')}"
            )

        except httpx.HTTPStatusError as e:
            result = f"查询失败：HTTP {e.response.status_code}，{e.response.text}"
        except httpx.TimeoutException:
            result = "查询超时，部署平台可能暂时不可用"
        except Exception as e:
            result = f"查询出错：{str(e)}"

    return [types.TextContent(type="text", text=result)]

async def _trigger_deploy(
    service_name: str,
    version: str,
    environment: str,
    reason: str
) -> list[types.TextContent]:
    """调用部署平台API触发部署"""

    # 安全检查：不允许部署到prod
    if environment == "prod":
        return [types.TextContent(
            type="text",
            text="[拒绝] prod环境部署需要在部署平台手动确认，无法通过API触发。"
        )]

    async with httpx.AsyncClient() as client:
        try:
            response = await client.post(
                f"{DEPLOY_API_BASE}/api/v1/deployments",
                json={
                    "service": service_name,
                    "version": version,
                    "environment": environment,
                    "reason": reason
                },
                headers={"Authorization": f"Bearer {DEPLOY_API_TOKEN}"},
                timeout=30.0
            )
            response.raise_for_status()
            data = response.json()

            result = (
                f"部署已触发！\n"
                f"部署ID：{data['deploy_id']}\n"
                f"服务：{service_name} → {environment}\n"
                f"版本：{version}\n"
                f"预计完成时间：{data.get('estimated_duration', '3-5分钟')}\n"
                f"查看进度：{DEPLOY_API_BASE}/deployments/{data['deploy_id']}"
            )

        except Exception as e:
            result = f"部署触发失败：{str(e)}"

    return [types.TextContent(type="text", text=result)]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "deploy-platform": {
      "command": "python3",
      "args": ["/path/to/your/deploy_mcp_server.py"],
      "env": {
        "DEPLOY_API_BASE": "https://deploy.internal.company.com",
        "DEPLOY_API_TOKEN": "${DEPLOY_TOKEN}"
      }
    }
  }
}

# 差的描述（太模糊）
description="Gets deployment info"

# 好的描述（明确触发场景）
description=(
    "查询指定服务的最新部署状态，包括部署时间、部署人、版本号、当前状态。"
    "当用户询问某个服务是否已上线、最新版本是什么、谁部署的时调用。"
    "不要用于查询代码本身，那应该用GitHub工具。"
)

{
  "mcpServers": {
    "github": {
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_TOKEN}"
      }
    }
  }
}

-- 为Claude专门创建只读账号
CREATE USER 'claude_readonly'@'localhost' IDENTIFIED BY '${password}';
GRANT SELECT ON myapp_prod.* TO 'claude_readonly'@'localhost';
-- 不要给INSERT、UPDATE、DELETE、DROP权限
FLUSH PRIVILEGES;

# 在自定义MCP Server里做脱敏
def mask_sensitive_fields(row: dict) -> dict:
    sensitive_keys = {'password', 'phone', 'id_card', 'bank_account', 'email'}
    return {
        k: '***MASKED***' if k.lower() in sensitive_keys else v
        for k, v in row.items()
    }

# 强制限制SELECT查询最多返回1000行
def safe_query(sql: str) -> str:
    sql = sql.strip().rstrip(';')
    if 'LIMIT' not in sql.upper():
        sql += ' LIMIT 100'
    return sql

import logging
from datetime import datetime

# 配置审计日志
audit_logger = logging.getLogger('mcp.audit')
audit_logger.setLevel(logging.INFO)
handler = logging.FileHandler('/var/log/mcp-audit.log')
handler.setFormatter(logging.Formatter('%(asctime)s %(message)s'))
audit_logger.addHandler(handler)

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    # 记录每一次工具调用
    audit_logger.info(json.dumps({
        "timestamp": datetime.now().isoformat(),
        "tool": name,
        "arguments": _redact_sensitive(arguments),  # 脱敏后记录
        "caller": "claude-code"
    }))

    # 正常处理...

# 查看今天所有trigger_deploy调用
jq 'select(.tool == "trigger_deploy")' /var/log/mcp-audit.log

# 统计今天数据库查询次数
jq 'select(.tool == "query") | .timestamp' /var/log/mcp-audit.log | wc -l

# .env文件（加到.gitignore）
GITHUB_TOKEN=ghp_xxxxxxxxxxxx
DB_PASSWORD=your_password_here
DEPLOY_TOKEN=your_token_here

# 加载.env后启动Claude
source .env && claude
# 或者用direnv自动加载：https://direnv.net/

# 直接在终端手动启动MCP Server，看有没有报错
npx @modelcontextprotocol/server-github

# 直接测试Token是否有效
curl -H "Authorization: Bearer $GITHUB_TOKEN" https://api.github.com/user

# 验证JSON语法
python3 -c "import json; json.load(open('.mcp.json')); print('JSON语法正确')"

# 确认文件在项目根目录
ls -la .mcp.json

# 重启Claude Code（配置文件在启动时读取，改了之后需要重启）

async def main():
    import sys
    print("MCP Server 启动中...", file=sys.stderr)
    async with stdio_server() as (read_stream, write_stream):
        print("等待连接...", file=sys.stderr)
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

# 安装MCP Inspector工具
npm install -g @modelcontextprotocol/inspector

# 启动Inspector，它会连接你的MCP Server并提供调试界面
mcp-inspector python3 /path/to/your_server.py