# mcp_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

mcp = Server("database-access")

@mcp.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="query_database",
            description="Execute SQL queries on the main database",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL query"}
                },
                "required": ["sql"]
            }
        )
    ]

@mcp.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "query_database":
        # 任何 MCP 客户端都可以调用这个功能
        result = execute_sql(arguments["sql"])
        return [TextContent(type="text", text=str(result))]

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

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import * as fs from "fs/promises";
import * as path from "path";

const server = new Server({
  name: "local-dev-tools",
  version: "1.0.0"
});

// 本地代码库分析
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "analyze-codebase") {
    const projectPath = request.params.arguments.path;

    // 直接访问本地文件系统
    const files = await fs.readdir(projectPath, { recursive: true });
    const codeFiles = files.filter(f => f.endsWith('.ts') || f.endsWith('.py'));

    // 并发读取文件内容（Skill 做不到这么高效）
    const contents = await Promise.all(
      codeFiles.map(f => fs.readFile(path.join(projectPath, f), 'utf-8'))
    );

    return {
      content: [{
        type: "text",
        text: JSON.stringify({ files: codeFiles, totalSize: contents.join('').length })
      }]
    };
  }
});

class AuthMiddleware:
    """MCP 服务端的统一权限网关"""

    async def __call__(self, request: ToolRequest) -> ToolResponse:
        # 1. 验证用户身份
        user = await self.authenticate(request.headers)

        # 2. 检查工具调用权限
        if not await self.check_permission(user, request.tool_name):
            raise PermissionDenied(f"User {user.id} cannot access {request.tool_name}")

        # 3. 速率限制
        if not await self.check_rate_limit(user):
            raise RateLimitExceeded("Too many requests")

        # 4. 执行并审计
        start_time = time.time()
        try:
            response = await self.execute_tool(request)
            await self.audit_log(user, request, response, duration=time.time() - start_time)
            return response
        except Exception as e:
            await self.audit_error(user, request, e)
            raise

@mcp.list_resources()
async def list_resources() -> list[Resource]:
    """动态发现所有可用的数据源"""
    resources = []

    # 数据库资源动态发现
    for db_config in await discover_databases():
        for table in await list_tables(db_config):
            resources.append(Resource(
                uri=f"db://{db_config.name}/{table}",
                name=f"{db_config.name}.{table}",
                description=f"Table {table} in {db_config.name}",
                mimeType="application/json"
            ))

    # 文件系统资源
    for root, dirs, files in os.walk("/shared/documents"):
        for file in files:
            resources.append(Resource(
                uri=f"file://{os.path.join(root, file)}",
                name=file,
                description=f"Document file",
                mimeType="text/plain"
            ))

    return resources

# AI 可以自主探索和调用这些资源
@mcp.read_resource()
async def read_resource(uri: str) -> str:
    if uri.startswith("db://"):
        return await query_table(uri)
    elif uri.startswith("file://"):
        return await read_file(uri)

@mcp.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "onboard_new_user":
        """跨系统自动化：新用户入职流程"""

        # 1. 在 HR 系统创建员工记录
        employee_id = await hr_api.create_employee(arguments)

        # 2. 在 AD 域创建账号
        ad_user = await ad_api.create_user({
            "username": arguments["email"].split("@")[0],
            "email": arguments["email"],
            "department": arguments["department"]
        })

        # 3. 在 Jira 创建工单分配设备
        ticket_id = await jira_api.create_ticket({
            "summary": f"设备申请: {arguments['name']}",
            "assignee": "it-support"
        })

        # 4. 在 Slack 发送欢迎消息
        await slack_api.send_message(
            channel="#announcements",
            text=f"欢迎新同事 {arguments['name']} 加入 {arguments['department']}！"
        )

        return [TextContent(type="text", text=f"""
            入职流程完成：
            - 员工ID: {employee_id}
            - AD账号: {ad_user.username}
            - 设备工单: {ticket_id}
        """)]

// 任何符合 MCP 标准的 Server 都可以被任何 MCP Client 使用
interface MCPTool {
  name: string;
  description: string;
  inputSchema: {
    type: "object";
    properties: Record<string, { type: string; description: string }>;
    required: string[];
  };
}

// 客户端可以动态加载任何 MCP Server
async function loadMCPServer(serverPath: string) {
  const server = spawn(serverPath);
  return new MCPClient(server.stdin, server.stdout);
}

// 生态示例：插件式架构
const pluginRegistry = {
  "filesystem": "/opt/mcp-servers/fileserver",
  "database": "/opt/mcp-servers/dbconnector",
  "cloud-storage": "/opt/mcp-servers/s3-connector"
};

// 用户可以随意替换某个插件，无需修改客户端代码

# SKILL.md 中的工作流定义
---
name: code-review
description: "Perform code review with standard checklist"
---

## Workflow

1. Read the code file provided by user
2. Check for:
   - Security vulnerabilities
   - Performance issues
   - Code style violations
   - Missing error handling
3. Generate review report in markdown format
4. Suggest specific fixes for each issue found

---
name: claude-code-optimization
description: "Claude Code specific workflow optimizations"
---

## Optimization

1. Use Edit tool instead of Write for existing files
2. Batch independent operations together
3. Use TodoWrite for task tracking
4. Follow Git Safety Protocol

---
name: code-review-with-git
description: "Code review that uses MCP Git tools for file operations"
---

## Workflow

1. 使用 MCP 工具获取当前 Git 仓库状态
2. 使用 MCP 工具读取变更的文件内容
3. 在本地执行代码审查分析
4. 生成审查报告
5. 使用 MCP 工具创建 Git commit（如果用户确认）