GenericMCPServerDevelopmentSkill mcp-server

这是一个用于构建MCP服务器的技能,提供通用架构、数据库集成、工具实现和测试模式,支持异步操作和企业级特性。

后端开发 0 次安装 0 次浏览 更新于 3/2/2026

以下是对提供的MCP服务器开发模式的中文翻译,包括核心服务器架构、数据库集成模式、工具实现模式、示例实现以及测试模式。

通用MCP服务器开发

这项技能提供了全面的模式和可重用代码,用于构建MCP(模型上下文协议)服务器,这些服务器可以将任何领域操作作为AI代理的工具暴露出来。遵循2025年最佳实践,涵盖性能、安全性、可维护性。

何时使用此技能

当您需要:

  • 为任何领域(不仅仅是待办事项)构建MCP服务器
  • 将数据库操作作为MCP工具暴露
  • 创建AI代理可访问的API
  • 实现异步MCP工具处理程序
  • 添加适当的错误处理和验证
  • 支持速率限制和缓存
  • 构建企业级MCP服务器
  • 与多个存储后端集成

1. 通用MCP服务器架构

# mcp_server/core.py
#!/usr/bin/env python3
"""
通用MCP服务器基础架构
提供任何MCP服务器实现的可重用模式
"""

import asyncio
import json
import logging
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Sequence, Union, Callable
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from enum import Enum

import redis.asyncio as redis
from mcp.server import Server, NotificationOptions, stdio
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
    Resource, Tool, TextContent, ImageContent, EmbeddedResource,
    LoggingLevel, CallToolRequest, EmptyResult,
    ListResourcesRequest, ListToolsRequest, ReadResourceRequest,
    GetPromptRequest, ListPromptsRequest
)
from pydantic import BaseModel, Field, validator
import aiofiles
import yaml
from pathlib import Path

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("mcp_server")

...(此处省略部分代码以节省篇幅)...

2. 数据库集成模式

# mcp_server/database.py
"""
通用数据库集成MCP服务器
支持多个ORM和连接模式
"""

import asyncio
from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional, TypeVar, Generic, Union
from datetime import datetime
import json

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, DateTime, Text, Boolean, select, update, delete, insert
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.pool import NullPool
import asyncpg
import motor.motor_asyncio
from redis.asyncio import Redis

...(此处省略部分代码以节省篇幅)...

3. 工具实现模式

# mcp_server/tools.py
"""
通用MCP工具实现模式
"""

from typing import Any, Dict, List, Optional, Union, Callable
from datetime import datetime, timedelta
import json
import uuid
from dataclasses import dataclass, field

from .core import BaseMCPServer, tool, RequestContext
from .database import DatabaseManager

...(此处省略部分代码以节省篇幅)...

4. 示例:构建通用任务管理MCP服务器

# examples/task_mcp_server.py
"""
示例:使用通用模式的任务管理MCP服务器
"""

import os
from typing import Dict, Any

from mcp_server.core import BaseMCPServer, ServerConfig, tool
from mcp_server.database import create_database_manager
from mcp_server.tools import CRUDBaseTool, BulkOperationTool

...(此处省略部分代码以节省篇幅)...

5. 测试模式

# tests/test_mcp_server.py
"""
通用MCP服务器测试模式
"""

import pytest
import asyncio
from typing import Dict, Any, List
from unittest.mock import Mock, AsyncMock

from mcp_server.core import BaseMCPServer, ServerConfig, RequestContext
from mcp_server.database import DatabaseManager, create_database_manager
from mcp_server.tools import CRUDBaseTool

...(此处省略部分代码以节省篇幅)...

这个通用的MCP服务器技能提供了一个可重用的基础,用于构建任何MCP服务器,不仅仅是待办事项。它包括:

  1. 核心服务器架构 - 具有速率限制、缓存和错误处理的基础类
  2. 数据库集成 - 支持PostgreSQL、MongoDB和SQLAlchemy的异步操作
  3. 工具模式 - 适用于任何实体的通用CRUD和批量操作模式
  4. 示例实现 - 展示如何使用通用模式构建任务管理服务器
  5. 测试框架 - 全面的测试模式和模拟

★ 洞察 ───────────────────────────────────── 这里的主要架构模式是MCP协议处理(BaseMCPServer)、数据访问层(DatabaseManager与多个后端)、业务逻辑层(CRUDBaseTool和BulkOperationTool)和特定实现(TaskMCPServer组合组件)之间的关注点分离。 这使得系统高度可重用和可维护。任何开发人员都可以通过定义他们的实体模式并组合通用工具来快速构建一个新的MCP服务器。 ─────────────────────────────────────────────────