MCP详解及手把手实战

from mcp.server import FastMCP
from selenium.webdriver import Remote, ChromeOptions
from selenium.webdriver.chromium.remote_connection import ChromiumRemoteConnection
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time
import mysql.connector
from mysql.connector import errorcode

mcp = FastMCP("server")
@mcp.tool()
def get_user_info(name: str):
    """
    根据用户名查询用户信息
    
    Args:
        name (str): 用户名
        
    Returns:
        dict: 用户信息，包含id, name, age, address字段
        None: 如果未找到用户
    """
    # 数据库连接参数
    db_config = {
        "host": "127.0.0.1",
        "port": 3306,
        "user": "root",  # 默认用户名，根据实际情况修改
        "password": "password",  # 根据实际情况修改
        "database": "test"
    }
    
    # 初始化连接变量
    conn = None
    cursor = None
    try:
        # 创建数据库连接
        conn = mysql.connector.connect(**db_config)
        # 创建游标
        cursor = conn.cursor(dictionary=True)
        # 查询SQL
        sql = "SELECT id, name, age, address FROM user WHERE name = %s"
        # 执行查询
        cursor.execute(sql, (name,))
        # 获取结果
        result = cursor.fetchone()
        return result
    except Exception as e:
        print(f"查询MySQL数据库出错: {e}")
        return None
    finally:
        # 关闭游标和连接
        if cursor:
            cursor.close()
        if conn:
            conn.close()
@mcp.tool()
def create_user(name: str, age: int, address: str):
    """
    创建新用户
    
    Args:
        name (str): 用户名
        age (int): 年龄
        address (str): 地址
        
    Returns:
        dict: 包含操作结果的字典
            - success (bool): 操作是否成功
            - message (str): 操作结果消息
            - user_id (int, 可选): 成功时返回的用户ID
    """
    # 数据库连接参数
    db_config = {
        "host": "127.0.0.1",
        "port": 3306,
        "user": "root",
        "password": "password",
        "database": "test"
    }
    
    # 初始化连接变量
    conn = None
    cursor = None
    try:
        # 创建数据库连接
        conn = mysql.connector.connect(**db_config)
        # 创建游标
        cursor = conn.cursor()
        # 检查用户名是否已存在
        check_sql = "SELECT COUNT(*) FROM user WHERE name = %s"
        cursor.execute(check_sql, (name,))
        count = cursor.fetchone()[0]
        if count > 0:
            return {
                "success": False,
                "message": f"用户名'{name}'已经存在"
            }
        
        # 插入新用户
        insert_sql = "INSERT INTO user (name, age, address) VALUES (%s, %s, %s)"
        cursor.execute(insert_sql, (name, age, address))
        # 提交事务
        conn.commit()
        # 获取新用户ID
        user_id = cursor.lastrowid
        return {
            "success": True,
            "message": "用户创建成功",
            "user_id": user_id
        }
    except Exception as e:
        print(f"创建用户出错: {e}")
        return {
            "success": False,
            "message": f"创建用户错误: {str(e)}"
        }
    finally:
        # 关闭游标和连接
        if cursor:
            cursor.close()
        if conn:
            conn.close()

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