from fastmcp import FastMCP

mcp = FastMCP("企业工具集")

# Tools：让 AI 执行操作
@mcp.tool
def query_user(user_id: str) -> dict:
    """查询用户信息"""
    return {"name": "张三", "role": "管理员"}

# Resources：为 AI 提供上下文
@mcp.resource("config://database")
def get_db_config() -> dict:
    """数据库配置"""
    return {"host": "localhost", "port": 5432}

# Prompts：预定义提示词模板
@mcp.prompt("customer_service")
def cs_prompt(issue_type: str) -> str:
    """客服提示词"""
    return f"专业处理{issue_type}问题：友好、准确、高效"

mcp.run()  # 启动服务，所有 AI 都能调用

# 定义工具
def query_user_openai(user_id: str):
    # OpenAI 格式的工具定义
    pass

def query_user_qwen(user_id: str):
    # Qwen 格式的工具定义 + 适配逻辑
    pass

# 还要写 100 多行的适配器...
response = qwen_tool_call_adapter(client.chat.completions.create(...))

from fastmcp import FastMCP

mcp = FastMCP("用户管理工具")

@mcp.tool
def query_user(user_id: str) -> dict:
    """查询用户信息 - 所有 AI 都能直接调用"""
    return database.get_user(user_id)

mcp.run()  # 一行启动，全 AI 兼容

from fastmcp import FastMCP

# 创建 MCP 服务器
mcp = FastMCP("我的第一个 AI 助手")

@mcp.tool
def say_hello(name: str) -> str:
    """向用户问好"""
    return f"你好，{name}！欢迎来到 MCP 的世界！"

if __name__ == "__main__":
    mcp.run()

import asyncio
from fastmcp import Client
from hello_server import mcp

async def test_hello():
    async with Client(mcp) as client:
        result = await client.call_tool("say_hello", {"name": "张三"})
        print(f"测试结果: {result.data}")

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

from fastmcp import FastMCP
import math

mcp = FastMCP("智能计算器")

@mcp.tool
def add(a: float, b: float) -> float:
    """计算两个数的和"""
    return a + b

@mcp.tool
def divide(a: float, b: float) -> float:
    """计算两个数的商（a 除以 b）"""
    if b == 0:
        raise ValueError("除数不能为零")
    return a / b

@mcp.tool
def sqrt(number: float) -> float:
    """计算平方根"""
    if number < 0:
        raise ValueError("负数没有实数平方根")
    return math.sqrt(number)

# 添加资源：提供上下文信息
@mcp.resource("calculator://help")
def get_help() -> str:
    """获取计算器使用帮助"""
    return """
    智能计算器使用说明：
    - add(a, b): 加法运算
    - divide(a, b): 除法运算（分母不能为零）
    - sqrt(number): 平方根运算（被开方数不能为负数）
    """

@mcp.resource("calculator://constants")
def get_constants() -> dict:
    """获取数学常数"""
    return {
        "pi": math.pi,
        "e": math.e,
        "golden_ratio": (1 + math.sqrt(5)) / 2
    }

if __name__ == "__main__":
    mcp.run()

# 错误写法
@mcp.tool
def bad_add(a, b):  # 没有类型提示
    return a + b

# 正确写法
@mcp.tool
def good_add(a: float, b: float) -> float:  # 完整类型提示
    return a + b

# 不好的描述
@mcp.tool
def calc(x: int, y: int) -> int:
    """计算"""  # 太模糊
    return x + y

# 好的描述
@mcp.tool
def add_integers(x: int, y: int) -> int:
    """计算两个整数的和"""  # 清楚明了
    return x + y

mcp.run()  # 适合与 Claude Desktop 直接连接

mcp.run(transport="http", port=8000)  # 适合 Web 集成

from fastmcp import FastMCP
import sqlite3

mcp = FastMCP("企业用户管理系统")

# 初始化数据库表结构
def init_database():
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            email TEXT UNIQUE NOT NULL,
            name TEXT NOT NULL,
            status TEXT DEFAULT 'active'
        )
    ''')
    conn.commit()
    conn.close()

init_database()

from typing import Dict

@mcp.tool
def get_user_by_id(user_id: int) -> Dict:
    """根据 ID 查询用户信息"""
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    cursor.execute(
        'SELECT id, username, email, name, status FROM users WHERE id = ?',
        (user_id,)
    )
    result = cursor.fetchone()
    conn.close()
    
    if result:
        return {
            'id': result[0], 'username': result[1], 'email': result[2],
            'name': result[3], 'status': result[4]
        }
    else:
        raise ValueError(f"未找到 ID 为 {user_id} 的用户")

@mcp.tool
def search_users(keyword: str) -> list:
    """根据关键词搜索用户"""
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    cursor.execute(
        'SELECT id, username, email, name, status FROM users '
        'WHERE username LIKE ? OR name LIKE ? OR email LIKE ?',
        (f'%{keyword}%', f'%{keyword}%', f'%{keyword}%')
    )
    results = cursor.fetchall()
    conn.close()
    
    return [
        {'id': r[0], 'username': r[1], 'email': r[2], 'name': r[3], 'status': r[4]}
        for r in results
    ]

@mcp.tool
def create_user(username: str, email: str, name: str) -> Dict:
    """创建新用户"""
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    try:
        cursor.execute(
            'INSERT INTO users (username, email, name) VALUES (?, ?, ?)',
            (username, email, name)
        )
        user_id = cursor.lastrowid
        conn.commit()
        return {"id": user_id, "username": username, "status": "created"}
    except sqlite3.IntegrityError as e:
        raise ValueError(f"用户创建失败：{e}")
    finally:
        conn.close()

@mcp.resource("users://stats")
def get_user_stats() -> Dict:
    """获取用户统计信息"""
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    
    # 统计活跃用户数
    cursor.execute('SELECT COUNT(*) FROM users WHERE status = "active"')
    active_count = cursor.fetchone()[0]
    
    # 统计总用户数
    cursor.execute('SELECT COUNT(*) FROM users')
    total_count = cursor.fetchone()[0]
    
    conn.close()
    
    return {
        "active_users": active_count,
        "total_users": total_count
    }

@mcp.resource("system://health")
def get_system_health() -> Dict:
    """获取系统健康状态"""
    return {
        "status": "healthy",
        "database": "connected",
        "timestamp": datetime.now().isoformat()
    }

import requests

@mcp.tool
def send_welcome_email(user_id: int) -> Dict:
    """给新用户发送欢迎邮件"""
    user = get_user_by_id(user_id)
    
    email_data = {
        "to": user['email'],
        "subject": f"欢迎加入，{user['name']}！",
        "template": "welcome",
        "username": user['username']
    }
    
    # 生产环境中这里调用真实 API
    # response = requests.post("https://api.yourservice.com/send-email", json=email_data)
    
    return {
        "status": "sent",
        "recipient": user['email'],
        "message": f"欢迎邮件已发送至 {user['email']}"
    }

@mcp.tool
def validate_email_domain(email: str) -> Dict:
    """验证邮箱域名是否有效"""
    domain = email.split('@')[1] if '@' in email else None
    
    if not domain:
        return {"valid": False, "reason": "邮箱格式不正确"}
    
    # 这里可以调用域名验证 API
    return {
        "valid": True,
        "domain": domain,
        "security_score": "high"
    }

import csv
from datetime import datetime

@mcp.tool
def export_users_csv() -> Dict:
    """导出用户数据到 CSV 文件"""
    filename = f"users_export_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    
    # 获取所有用户数据
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    cursor.execute('SELECT id, username, email, name, status FROM users')
    users = cursor.fetchall()
    conn.close()
    
    # 写入 CSV
    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(['ID', '用户名', '邮箱', '姓名', '状态'])
        writer.writerows(users)
    
    return {
        "filename": filename,
        "record_count": len(users),
        "exported_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    }

@mcp.tool
def import_users_csv(file_path: str) -> Dict:
    """从 CSV 文件导入用户"""
    imported = 0
    failed = []
    
    with open(file_path, 'r', encoding='utf-8') as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            try:
                create_user(
                    username=row['username'],
                    email=row['email'],
                    name=row['name']
                )
                imported += 1
            except Exception as e:
                failed.append({"row": row, "error": str(e)})
    
    return {
        "imported": imported,
        "failed": len(failed),
        "details": failed if failed else None
    }

if __name__ == "__main__":
    mcp.run()  # 默认 STDIO 模式

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--http", action="store_true")
    args = parser.parse_args()
    
    if args.http:
        mcp.run(transport="http", port=8000)
    else:
        mcp.run()

from fastmcp import FastMCP
from fastmcp.server.auth import BearerAuthProvider
from fastmcp.server.auth.providers.bearer import RSAKeyPair

# 生成安全密钥和令牌
key_pair = RSAKeyPair.generate()
access_token = key_pair.create_token(
    audience="enterprise-system",
    expires_in=86400  # 24 小时有效
)

# 配置认证服务器
auth_provider = BearerAuthProvider(
    public_key=key_pair.public_key,
    audience="enterprise-system"
)

mcp = FastMCP("AI 智能系统", auth=auth_provider)

@mcp.tool
def secure_query(query: str) -> dict:
    """安全查询 - 需要认证"""
    return {"data": "only_authorized_access", "query": query}

from fastmcp import Middleware
from collections import defaultdict, deque
import time
import logging

logger = logging.getLogger(__name__)

# 1. 日志中间件 - 记录一切
class LoggingMiddleware(Middleware):
    async def process_request(self, context, next_handler):
        start_time = time.time()
        
        try:
            result = await next_handler()
            duration = time.time() - start_time
            logger.info(f"请求成功: {context.tool_name}, 耗时: {duration:.3f}s")
            return result
        except Exception as e:
            logger.error(f"请求失败: {context.tool_name}, 错误: {e}")
            raise

# 2. 限流中间件 - 防止攻击
class RateLimitMiddleware(Middleware):
    def __init__(self, max_requests=100, time_window=60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.request_counts = defaultdict(deque)
    
    async def process_request(self, context, next_handler):
        client_id = context.client_id or "anonymous"
        now = time.time()
        
        # 清理过期请求记录
        while self.request_counts[client_id] and \
              self.request_counts[client_id][0] < now - self.time_window:
            self.request_counts[client_id].popleft()
        
        # 检查频率限制
        if len(self.request_counts[client_id]) >= self.max_requests:
            raise ValueError("请求频率过高，请稍后再试")
        
        self.request_counts[client_id].append(now)
        return await next_handler()

# 3. 缓存中间件 - 提升性能
class SmartCacheMiddleware(Middleware):
    def __init__(self, ttl=600):
        self.ttl = ttl
        self.cache = {}
    
    async def process_request(self, context, next_handler):
        # 只对读操作缓存
        if context.tool_name.startswith("get_") or context.tool_name.startswith("query_"):
            cache_key = f"{context.tool_name}:{hash(str(context.params))}"
            
            if cache_key in self.cache:
                cached_result, timestamp = self.cache[cache_key]
                if time.time() - timestamp < self.ttl:
                    return cached_result
            
            result = await next_handler()
            self.cache[cache_key] = (result, time.time())
            return result
        
        return await next_handler()

# 应用所有中间件
mcp.add_middleware(LoggingMiddleware())
mcp.add_middleware(RateLimitMiddleware(max_requests=200))
mcp.add_middleware(SmartCacheMiddleware(ttl=600))

# 主控服务器 - 业务编排
main_server = FastMCP("企业业务系统")

# 导入各业务模块
user_service = FastMCP("用户服务")
order_service = FastMCP("订单服务")
notify_service = FastMCP("通知服务")

async def setup_services():
    await main_server.import_server(user_service, prefix="user")
    await main_server.import_server(order_service, prefix="order")
    await main_server.import_server(notify_service, prefix="notify")

# 业务流程编排
@main_server.tool
async def complete_order(user_id: int, items: list) -> dict:
    """完整的订单处理流程"""
    try:
        # 1. 验证用户 -> 2. 创建订单 -> 3. 发送通知
        user = await main_server.call_tool("user_get_user", {"user_id": user_id})
        order = await main_server.call_tool("order_create", {"user_id": user_id, "items": items})
        await main_server.call_tool("notify_email", {
            "to": user["email"],
            "subject": "订单创建成功",
            "content": f"您的订单 #{order['id']} 已创建成功"
        })
        
        return {"status": "success", "order": order}
    except Exception as e:
        logger.error(f"订单处理失败: {e}")
        return {"status": "error", "message": str(e)}

from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache
import asyncpg
from contextlib import asynccontextmanager

class AsyncMCPServer:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=20)
        self.db_pool = None
    
    @asynccontextmanager
    async def lifespan(self, server):
        # 启动时初始化连接池
        self.db_pool = await asyncpg.create_pool(
            "postgresql://user:pass@localhost/db",
            min_size=5, max_size=20
        )
        
        try:
            yield
        finally:
            await self.db_pool.close()

# 智能缓存 - 减少数据库压力
@lru_cache(maxsize=1000)
def get_cached_result(key: str, timestamp: int):
    # 基于时间戳的缓存
    return None

# 高性能查询
@mcp.tool
async def fast_user_query(user_id: int) -> dict:
    """高性能用户查询 - 带缓存"""
    # 先检查缓存，再查数据库
    cache_key = f"user:{user_id}"
    cached = get_cached_result(cache_key, int(time.time() // 300))
    
    if cached:
        return cached
    
    # 异步数据库查询
    async with db_pool.acquire() as conn:
        row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
        result = dict(row) if row else None
        
        # 更新缓存
        if result:
            get_cached_result.cache_clear()
        
        return result

from dataclasses import dataclass
import psutil

@dataclass
class PerformanceMetrics:
    """性能指标数据类"""
    timestamp: float
    cpu_percent: float
    memory_percent: float
    avg_response_time: float
    request_count: int

class MetricsCollector:
    """性能指标收集器"""
    def __init__(self):
        self.request_count = 0
        self.request_times = []
    
    def record_request(self, duration: float):
        self.request_count += 1
        self.request_times.append(duration)
        # 只保留最近 1000 条记录
        if len(self.request_times) > 1000:
            self.request_times = self.request_times[-1000:]
    
    def collect_metrics(self) -> PerformanceMetrics:
        avg_response = sum(self.request_times) / len(self.request_times) if self.request_times else 0
        
        return PerformanceMetrics(
            timestamp=time.time(),
            cpu_percent=psutil.cpu_percent(),
            memory_percent=psutil.virtual_memory().percent,
            avg_response_time=avg_response,
            request_count=self.request_count
        )

# 告警管理
class AlertManager:
    def __init__(self):
        self.alert_rules = []
    
    def add_rule(self, rule, message, level="warning"):
        self.alert_rules.append({"rule": rule, "message": message, "level": level})
    
    def check_alerts(self, metrics: PerformanceMetrics):
        for rule_config in self.alert_rules:
            if rule_config["rule"](metrics):
                self._send_alert(rule_config)
    
    def _send_alert(self, config):
        level = config["level"]
        message = config["message"]
        print(f"🚨 [{level.upper()}] {message}")
        # 这里可以集成邮件、短信、钉钉等通知渠道

# 配置告警规则
alert_manager = AlertManager()
alert_manager.add_rule(
    rule=lambda m: m.cpu_percent > 90,
    message="CPU 使用率超过 90%，请检查系统负载",
    level="critical"
)
alert_manager.add_rule(
    rule=lambda m: m.memory_percent > 85,
    message="内存使用率超过 85%，建议优化内存使用",
    level="warning"
)
alert_manager.add_rule(
    rule=lambda m: m.avg_response_time > 5,
    message="平均响应时间超过 5 秒，可能存在性能瓶颈",
    level="warning"
)

@mcp.tool
def get_system_health() -> dict:
    """获取系统健康状态"""
    metrics = collector.collect_metrics()
    alert_manager.check_alerts(metrics)
    
    issues = []
    if metrics.cpu_percent > 80:
        issues.append("CPU 使用率偏高")
    if metrics.memory_percent > 85:
        issues.append("内存使用率偏高")
    if metrics.avg_response_time > 3:
        issues.append("响应时间较长")
    
    return {
        "status": "unhealthy" if issues else "healthy",
        "issues": issues,
        "metrics": {
            "cpu_percent": round(metrics.cpu_percent, 2),
            "memory_percent": round(metrics.memory_percent, 2),
            "avg_response_time": round(metrics.avg_response_time, 3),
            "request_count": metrics.request_count
        },
        "timestamp": datetime.fromtimestamp(metrics.timestamp).isoformat()
    }

# enterprise_server.py - 生产级 MCP 服务器
from fastmcp import FastMCP
from fastmcp.server.auth import BearerAuthProvider
from fastmcp.server.auth.providers.bearer import RSAKeyPair
import asyncio
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 生成认证密钥
key_pair = RSAKeyPair.generate()
access_token = key_pair.create_token(
    audience="enterprise-mcp",
    expires_in=86400  # 24 小时
)

# 创建企业级 MCP 服务器
mcp = FastMCP(
    name="企业级 MCP 智能系统",
    auth=BearerAuthProvider(public_key=key_pair.public_key)
)

# 应用中间件栈
mcp.add_middleware(LoggingMiddleware())
mcp.add_middleware(RateLimitMiddleware(max_requests=200))
mcp.add_middleware(SmartCacheMiddleware(ttl=600))

# 初始化监控
collector = MetricsCollector()
alert_manager = AlertManager()

# 配置告警规则
alert_manager.add_rule(
    rule=lambda m: m.cpu_percent > 90,
    message="CPU 使用率超过 90%",
    level="critical"
)

# 启动监控任务
async def startup_monitoring():
    while True:
        try:
            metrics = collector.collect_metrics()
            alert_manager.check_alerts(metrics)
            await asyncio.sleep(60)  # 每分钟检查
        except Exception as e:
            logger.error(f"监控任务异常: {e}")

if __name__ == "__main__":
    print("🏢 企业级 MCP 服务器启动中...")
    print(f"🔐 访问令牌: {access_token}")
    print("⚡ 已启用特性: 认证、日志、限流、缓存、监控、告警")
    
    # 启动监控
    asyncio.create_task(startup_monitoring())
    
    # 启动服务器
    mcp.run(transport="http", host="0.0.0.0", port=8000)