AI 如何与我的数据库对话？MySQL 和 Gemini

爱可生开源社区

发布于 2025-07-26 12:46:31

21000

代码可运行

文章被收录于专栏：爱可生开源社区爱可生开源社区

运行总次数：0

代码可运行

作者：Fernando Laudares Camargos，Senior Support Engineer, Percona

原文：https://www.percona.com/blog/how-can-ai-talk-to-my-database-part-two-mysql-and-gemini/，July 18, 2025

爱可生开源社区翻译，本文约 3100 字，预计阅读需要 10 分钟。

我最初的实验是使用 FastMCP Python 框架以及 Anthropic 和 OpenAI 的 API 创建 MCP 服务器，以便 AI 访问 PostgreSQL 数据库。

实验中，我遇到了一个重要的需求：目前，这两个 API 只能通过 HTTPS 协议通过公共 URL 与 MCP 服务器通信。在研究如何实现这一点的过程中（这促成了我在本系列的第一篇博文《AI 如何与我的数据库对话：PostgreSQL》[1]），我注意到 Gemini 的 API 有所不同：它可以访问本地 MCP 服务器。

因此，我决定尝试一下，但这次使用 MySQL 作为目标数据库。

使用 FastMCP 创建 MySQL MCP 服务器

正如我在第一篇博客文章中解释的那样， FastMCP 框架[2]为我们完成了大部分工作。在最简单的实现中，我们需要做的就是指示如何访问我们的 MySQL 数据库：

import json
import mysql.connector
from fastmcp import FastMCP

mcp = FastMCP(name="MySQL demo")

@mcp.tool()
def query_data(sql_query: str) -> str:
'''Execute MySQL queries safely for the actors table inside the test database and return structured JSON response.'''

  DB_NAME = "sakila"
  DB_USER = "mcpreader"
  DB_PASS = "secret"
  DB_HOST = "localhost"
  DB_PORT = "3306"

  conn = None
  cursor = None

try:
    conn = mysql.connector.connect(
      database=DB_NAME,
      user=DB_USER,
      password=DB_PASS,
      host=DB_HOST,
      port=DB_PORT
    )
    cursor = conn.cursor()
    cursor.execute(sql_query)
    rows = cursor.fetchall()
finally:
    if cursor isnotNone:
      cursor.close()
    if conn isnotNone:
      conn.close()
return json.dumps(rows, indent=2)

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

与第一篇博文中使用的 PostgreSQL MCP 服务器代码相比，改动很小：我们基本上是用 MySQL 驱动程序替换了 psycopg2 （PostgreSQL 的 Python 驱动程序）。我将这段代码保存在名为 mcp-mysql.py 的文件中。

MySQL 目标数据库

这次，我在我的 Ubuntu 机器上安装了一个 MySQL 测试数据库，我还将在那里运行 MCP 服务器并测试 Gemini 集成：

sudo apt-get install mysql-server-8.0

这次，我将不再像之前那样使用简单的两行 actors 表，而是利用 Sakila 示例数据库，为 AI 提供更多数据。下载其源码包，解压 tarball：

cd /tmp
curl -O https://downloads.mysql.com/docs/sakila-db.tar.gz
tar zxvf sakila-db.tar.gz

并将模式和数据加载到 MySQL 中：

$ sudo mysql

然后：

mysql> source /tmp/sakila-db/sakila-schema.sql;
mysql> source /tmp/sakila-db/sakila-data.sql;

最后，让我们为 MCP 服务器创建一个只读用户来连接到新的 sakila 数据库：

mysql> create user 'mcpreader'@'localhost' identified by 'secret';
mysql> grant select on sakila.* to 'mcpreader'@'localhost';

运行 MCP 服务器

正如第一篇博文中解释的那样，我们可以使用 uv[3] 创建一个虚拟环境来安装必要的 Python 模块（FastMCP 和 Python 的 MySQL 连接器）并运行 MCP 服务器，这次使用 sse 传输方法而不是 http。

curl -LsSf https://astral.sh/uv/install.sh | sh
uv venv e1
source e1/bin/activate
uv pip install fastmcp mysql-connector-python
fastmcp run -t sse mcp-mysql.py

您应该会看到类似以下内容的消息，表明服务器已启动：

[07/07/25 17:06:57] INFO     Starting MCP server 'MySQL demo' with transport 'sse' on                 server.py:1429
                             http://127.0.0.1:8000/sse/

测试数据库集成

让我们连接到同一个虚拟环境并启动 FastMCP 客户端来验证我们的 MCP 服务器是否按预期工作（您需要从启动虚拟环境的同一目录运行下面的源命令）：

source e1/bin/activate
ipython3

进入 Python 环境后，打开与 MCP 服务器的连接并请求其工具列表：

from fastmcp import Client
client = Client("http://127.0.0.1:8000/sse/")
async with client:
  tools = await client.list_tools()

如果一切顺利，您应该会看到我们定义的 query_data 工具及其描述：

In [4]: print(tools)
[Tool(name='query_data', title=None, description='Execute MySQL queries safely for the actors table inside the test database and return structured JSON response.', inputSchema={'properties': {'sql_query': {'title': 'Sql Query', 'type': 'string'}}, 'required': ['sql_query'], 'type': 'object'}, outputSchema={'properties': {'result': {'title': 'Result', 'type': 'string'}}, 'required': ['result'], 'title': '_WrappedResult', 'type': 'object', 'x-fastmcp-wrap-result': True}, annotations=None, meta=None)]

当您配置 AI API 访问 MCP 服务器以获取可用工具列表以及每个工具提供的内容的描述时，大多数 AI API 都会调用 list_tools() 函数，这对于 AI 了解在何种情况下可以使用该工具至关重要。

让我们尝试一下我们的工具：

async with client:
  result = await client.call_tool("query_data", {"sql_query": "select count(*) from actor"})

然后检查结果：

In [8]: print(result.structured_content['result'])
[
  [
    200
  ]
]

它正在发挥作用。

访问数据库的方法，Gemini

让我们使用不同的虚拟环境来测试 Gemini 与 FastMCP 的集成[4]:

uv venv e2
source e2/bin/activate

在启动 ipython3 之前，我们需要安装 google-genai 和 fastmcp Python 模块并提供 Gemini API 密钥（您需要使用您的 Google 帐户创建）作为环境变量：

uv pip install fastmcp google-genai
export GEMINI_API_KEY="<redacted>"
ipython3

在 Python 环境中，我们需要实例化连接到本地 MCP 服务器的 FastMCP 客户端和 Gemini 客户端：

from fastmcp import Client
from google import genai

mcp_client = Client("http://127.0.0.1:8000/sse/")
gemini_client = genai.Client()

这次，我想利用多表 Sakila 数据库并向 AI 询问一些需要通过多连接查询来回答的问题：

question = "Which films feature Morgan McDormand and are suitable for a general audience?"

完整请求的制定方法如下：

async with mcp_client:
   response = await gemini_client.aio.models.generate_content(
     model="gemini-2.0-flash",
     contents=question,
     config=genai.types.GenerateContentConfig(
     temperature=0,
     tools=[mcp_client.session]
   )
 )

请注意，我们通过向 Gemini 提供 FastMCP 客户端会话来让它使用 MCP 服务器中的工具。以下是它的完整初始回复：

In [6]: print(response)
sdk_http_response=HttpResponse(
  headers=<dict len=11>
) candidates=[Candidate(
  avg_logprobs=-0.079243004322052,
  content=Content(
    parts=[
      Part(
        text="""I cannot provide information about film ratings or suitability for specific audiences. However, I can search for films featuring Morgan McDormand. Do you want me to proceed with that?
"""
      ),
    ],
    role='model'
  ),
  finish_reason=<FinishReason.STOP: 'STOP'>
)] create_time=None response_id=None model_version='gemini-2.0-flash' prompt_feedback=None usage_metadata=GenerateContentResponseUsageMetadata(
  candidates_token_count=36,
  candidates_tokens_details=[
    ModalityTokenCount(
      modality=<MediaModality.TEXT: 'TEXT'>,
      token_count=36
    ),
  ],
  prompt_token_count=44,
  prompt_tokens_details=[
    ModalityTokenCount(
      modality=<MediaModality.TEXT: 'TEXT'>,
      token_count=44
    ),
  ],
  total_token_count=80
) automatic_function_calling_history=[] parsed=None

我仍然需要找到更好地处理这些输出的方法，但这里有一个目前可用的快速技巧：

In [7]: d = response.to_json_dict()
In [8]: print(d['candidates'][0]['content']['parts'][0]['text'])

I cannot provide information about film ratings or suitability for specific audiences. However, I can search for films featuring Morgan McDormand. Do you want me to proceed with that?

我尝试说得更具体一些：

【我】：在本地的 sakila 数据库中，您可以找到哪些适合大众观看的摩根·麦克多蒙德主演的电影？

对此的回应是：

【AI】：我无法提供电影分级或特定观众适宜性的信息。不过，我可以搜索摩根·麦克多蒙德主演的电影。你想让我继续吗？

也许我应该 提高 “温度（temperature）” 设置[5] ，因为“较低的温度适合那些需要更确定性或更不开放的反应的提示，而较高的温度可以带来更多样化或更有创意的结果。”

让我们尝试一下，同时强调它实际上可以提供我所要求的信息：

async with mcp_client:
   response = await gemini_client.aio.models.generate_content(
     model="gemini-2.0-flash",
     contents="You should be able to find the answer to my question if you query the sakila database",
     config=genai.types.GenerateContentConfig(
     temperature=0.1,
     tools=[mcp_client.session]
   )
 )

答复如下：

【AI】：好的，我明白了。我会使用 query_data 函数查询 sakila 数据库来找到您问题的答案。请继续提问。

耶，看来我们快到了。我再试一次：

【我】：摩根·麦克多蒙德主演的哪些电影适合大众观众观看？

【AI】：我无法回答这个问题，因为现有的工具缺乏根据演员和观众适合度筛选电影的能力。

没办法。我期望它生成的查询结果如下：

$ mysql --user=mcpreader --password=secret --host=localhost --database=sakila -e "select film.title from film inner join film_actor on film.film_id=film_actor.film_id inner join actor on film_actor.actor_id=actor.actor_id where film.rating='G' and actor.first_name='MORGAN' and actor.last_name='MCDORMAND'"
+--------------------+
| title              |
+--------------------+
| DIVORCE SHINING    |
| SAGEBRUSH CLUELESS |
+--------------------+

在美国电影协会的电影分级制度[6]中，“G”代表普通观众。但人工智能怎么可能知道这些呢？

那么，让我们尝试一些更简单的事情：

【我】：我们在 sakila 数据库中能找到多少位演员？

【AI】：我无法直接访问 sakila 数据库。但是，我可以查询测试数据库中的 actors 表。请问可以这样做吗？

没有“ 测试” 数据库。哦，等等！它不知道自己连接到了 sakila。它本可以检查一下：

$ mysql --user=mcpreader --password=secret --host=localhost --database=sakila -e "select database()"
+------------+
| database() |
+------------+
| sakila     |
+------------+

【我】：是的，请这样做。

【AI】：我准备好了。请提供您要执行的 SQL 查询。

嗯……我没想到会这样。现在我只想看看它是否真的可以访问我的测试数据库：

【我】：好的，运行查询：select count(*) from actor;

【AI】：表中有 200 名演员。

嗯，差不多就是这样。干得好……

不同的大模型将提供不同的结果

在我将这篇文章提交给同事审阅后，在发表之前，他们建议我使用的 LLM 模型可能不合适。到目前为止，我与 Gemini 的交互都是使用 gemini-2.0-flash 模型进行的，Vadim 建议我尝试更高级的 2.0 甚至 2.5-pro 模型。我决定直接使用 pro 模型，去掉了温度变量，并重复了我最初的问题：

question = "Which films feature Morgan McDormand and are suitable for a general audience?"

async with mcp_client:
  response = await gemini_client.aio.models.generate_content(
    model="gemini-2.5-pro",
    contents=question,
    config=genai.types.GenerateContentConfig(
    tools=[mcp_client.session]
  )
)

这一次，AI 花了更多时间处理请求，但却给出了肯定的答复：

【AI】：摩根·麦克多蒙德主演过以下适合大众观看的电影：

DIVORCE SHINING
SAGEBRUSH CLUELESS

这真是太棒了！它从一开始就做对了。以下是它使用的流程分解：

text="""I can help with that. First, I need to check if I can find the actor in the database. Then, I will search for the films they have been in and filter them by audience rating.
I will start by searching for the actor "Morgan McDormand"."""

sql_query: "SELECT * FROM actors WHERE full_name = 'Morgan McDormand'"

text='I apologize, it seems there was an issue with the table name. Let me try again with a different table name. It's possible the table is named `actor` (singular) instead of `actors`. I will also broaden my search to include similar names, in case "Morgan McDormand" is not in the database.'

sql_query: "SELECT * FROM actor WHERE first_name = 'MORGAN' AND last_name = 'MCDORMAND'"

text="It seems I'm having trouble retrieving the data. The error indicates a data type that cannot be handled correctly. I will try to select only the essential information to avoid this issue. Let me try again to find the actor 'Morgan McDormand'."

sql_query: "SELECT actor_id, first_name, last_name FROM actor WHERE first_name = 'MORGAN' AND last_name = 'MCDORMAND'"

text="Great, I have found the actor ID for Morgan McDormand. Now I will search for the films they have starred in that have a 'G' rating, which is suitable for a general audience."

sql_query: "SELECT T2.title, T2.rating FROM film_actor AS T1 INNER JOIN film AS T2 ON T1.film_id = T2.film_id WHERE T1.actor_id = 114 AND T2.rating = 'G'"

我仍然不明白为什么它仍然试图猜测表结构而不是使用查询来获取正确的详细信息，但这没关系.

我忘了提一个细节：一开始我有点怀疑 AI 实际运行的查询可能比它宣传的要多，所以我在测试服务器中启用了通用日志。注意每次查询执行间隔的秒数：

2025-07-10T19:18:02.130803Z    40 Connect mcpreader@localhost on sakila using SSL/TLS
2025-07-10T19:18:02.145585Z    40 Query SET NAMES 'utf8mb4' COLLATE 'utf8mb4_0900_ai_ci'
2025-07-10T19:18:02.153738Z    40 Query set autocommit=0
2025-07-10T19:18:02.158572Z    40 Query SELECT * FROM actors WHERE full_name = 'Morgan McDormand'
2025-07-10T19:18:02.205956Z    40 Quit 
2025-07-10T19:18:08.495421Z    41 Connect mcpreader@localhost on sakila using SSL/TLS
2025-07-10T19:18:08.496699Z    41 Query SET NAMES 'utf8mb4' COLLATE 'utf8mb4_0900_ai_ci'
2025-07-10T19:18:08.497367Z    41 Query set autocommit=0
2025-07-10T19:18:08.498485Z    41 Query SELECT * FROM actor WHERE first_name = 'MORGAN' AND last_name = 'MCDORMAND'
2025-07-10T19:18:08.506876Z    41 Quit 
2025-07-10T19:18:22.409341Z    42 Connect mcpreader@localhost on sakila using SSL/TLS
2025-07-10T19:18:22.410829Z    42 Query SET NAMES 'utf8mb4' COLLATE 'utf8mb4_0900_ai_ci'
2025-07-10T19:18:22.411516Z    42 Query set autocommit=0
2025-07-10T19:18:22.411987Z    42 Query SELECT actor_id, first_name, last_name FROM actor WHERE first_name = 'MORGAN' AND last_name = 'MCDORMAND'
2025-07-10T19:18:22.413449Z    42 Quit 
2025-07-10T19:18:30.703746Z    43 Connect mcpreader@localhost on sakila using SSL/TLS
2025-07-10T19:18:30.704448Z    43 Query SET NAMES 'utf8mb4' COLLATE 'utf8mb4_0900_ai_ci'
2025-07-10T19:18:30.705126Z    43 Query set autocommit=0
2025-07-10T19:18:30.705730Z    43 Query SELECT T2.title, T2.rating FROM film_actor AS T1 INNER JOIN film AS T2 ON T1.film_id = T2.film_id WHERE T1.actor_id = 114 AND T2.rating = 'G'
2025-07-10T19:18:30.718344Z    43 Quit

除此之外，它不是“炫酷”的，而是更“专业”（高效）的。我今天学到的教训是：如果你不想在人工智能上浪费时间，那就明智地选择你的模型吧。😉

参考资料

[1]

AI 如何与我的数据库对话第一部分: https://www.percona.com/blog/how-can-ai-talk-to-my-postgresql-database/

[2]

FastMCP: https://github.com/jlowin/fastmcp

[3]

uv: https://docs.astral.sh/uv/

[4]

集成 Gemini: https://gofastmcp.com/integrations/gemini

[5]

temperature 设置: https://ai.google.dev/gemini-api/docs/prompting-strategies