使用说明
MCP（模型上下文协议）+Gemini AI+车辆API集成
📋 概述
该项目展示了一个三层架构，连接：
- 谷歌双子座人工智能 -理解用户查询并决定调用哪个工具的语言模型
- MCP服务器 -提供不断增长的工具包（查找、搜索、摘要和更新）的协议服务器
- 数据源API -带有车辆信息的REST API，现在带有搜索和更新端点
代理人现在不仅可以 阅读 按ID从数据源获取，但也 按条件搜索车辆, 获取库存摘要，甚至 更改车辆状态，将其变成一个更自主的工具集。 该系统的工作原理是让Gemini AI决定使用哪些工具，MCP服务器桥接通信，数据源提供实际数据。
🏗️ 建筑
┌─────────────────┐
│ User Query │
│ (mcp_client.py) │
└────────┬────────┘
│
├─────────────────────────────┐
│ │
┌────▼──────────────────────────┐ │
│ Google Gemini AI │ │
│ (2.5-flash-lite model) │ │
│ │ │
│ - Understands user intent │ │
│ - Decides which tools to use │ │
│ - Generates final response │ │
└────┬──────────────────────────┘ │
│ │
│ 1. Sends tools definitions │
│ 2. Gets back tool request │
│ 3. Sends tool result │
│ 4. Gets final answer │
│ │
┌────▼──────────────────────────┐ │
│ MCP Server (mcp_server.py) │ │
│ │ │
│ - Runs as subprocess │ │
│ - Exposes get_vehicle_details │ │
│ - Communicates via stdio │ │
└────┬──────────────────────────┘ │
│ │
│ HTTP GET Request │
│ │
┌────▼──────────────────────────┐ │
│ Data Source API (data_source) │ │
│ (FastAPI on port 8001) │ │
│ │ │
│ - /vehicles/{vehicle_id} │ │
│ - /vehicles/search │ │
│ - PATCH /vehicles/{vehicle_id}│ │
│ - Returns: make, model, │ │
│ status, destination │ │
└───────────────────────────────┘ │
│ │
└─────────────────────────────┘
📂 文件分解
1. data_source.py -数据提供者
目的：为车辆数据提供服务的REST API
框架：FastAPI+Uvicorn
端口: 8001
它的作用：
- 提供车辆（101、102等）的模拟数据库
- 显示端点：
GET /vehicles/{vehicle_id}用于直接查找GET /vehicles/search基于标准的语义搜索PATCH /vehicles/{vehicle_id}更改状态
- 返回JSON格式：
make,model,status,destination
代码流：
VEHICLES = {
"101": {"make": "Toyota", "model": "HiAce", ...},
"102": {"make": "Honda", "model": "Fit", ...},
}
@app.get("/vehicles/{vehicle_id}") # Endpoint
async def get_vehicle(vehicle_id: str):
if vehicle_id not in VEHICLES:
raise HTTPException(status_code=404, ...)
return VEHICLES[vehicle_id] # Returns vehicle data as JSON
要运行：
python data_source.py
# Server runs on http://127.0.0.1:8001
# Endpoints available:
# GET /vehicles/{id}
# GET /vehicles/search?status=Shipped&make=Toyota
# POST /vehicles (JSON body with make/model/status/destination)
# PATCH /vehicles/{id}?status=In+Port
2. mcp_server.py -工具提供商
目的：将数据源API包装为工具的MCP服务器
框架：FastMCP（基于MCP协议构建）
运输：stdio（通过stdin/stdout进行通信）
它的作用：
- 当mcp_client.py运行时，作为子进程启动
- 提供多种工具:
get_vehicle_details,search_vehicles,inventory_summary，以及change_status - 从数据源API（localhost:8001）获取数据
- 将HTTP响应转换为具有丰富错误消息的工具输出
代码流：
@mcp.tool() # Defines a callable tool
async def get_vehicle_details(vehicle_id: str) -> str:
"""Description of what this tool does"""
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(f"{DATA_SOURCE_URL}/vehicles/{vehicle_id}")
if response.status_code == 200:
data = response.json()
# Format result for Gemini
return f"Vehicle {vehicle_id}: {data['make']} {data['model']} is currently {data['status']} heading to {data['destination']}."
return f"Error: Could not find details for vehicle {vehicle_id}."
if __name__ == "__main__":
mcp.run(transport="stdio") # Communicates via stdin/stdout
主要特点：
- 错误处理：当事情失败时，工具会返回描述性消息和建议
- 超时：5秒限制，防止悬挂
- 模式：根据函数参数自动生成输入模式
- 代理工具:
search_vehicles当ID未知时，充当代理的眼睛inventory_summary按状态/目的地提供宏观级别计数change_status允许代理执行写入操作
手动运行（用于测试）：
python mcp_server.py
# Listens on stdin, outputs to stdout
3. mcp_client.py -编排者和链经理
目的：Gemini AI和MCP工具之间协调的主要应用程序
框架：MCP客户端SDK+谷歌Genai SDK
它的作用：
- 将MCP服务器作为子进程启动
- 连接到Gemini AI API
- 从MCP服务器获取可用工具（包括新的搜索/摘要/更新功能）
- 将工具与用户查询一起发送到Gemini
- 处理Gemini的工具请求 循环，链接多个调用
- 通过MCP服务器执行工具并记录原始输出以进行调试
- 将结果发送回Gemini，直到生成纯文本答案
- 妥善处理API错误（如配额限制）
代码流：
步骤1：初始化MCP服务器连接
server_params = StdioServerParameters(
command="python",
args=["mcp_server.py"] # Start mcp_server.py as subprocess
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize() # Handshake with server
第二步：获取可用工具
mcp_tools = await session.list_tools() # Returns available tools from MCP server
# Result: [Tool(name="get_vehicle_details", description="...", inputSchema={...})]
步骤3：为Gemini转换工具
gemini_tools = [
types.Tool(
function_declarations=[
{
"name": tool.name, # "get_vehicle_details"
"description": tool.description, # Tool description
"parameters": clean_schema(tool.inputSchema), # Input parameters
} for tool in mcp_tools.tools
]
)
]
# clean_schema() removes "additionalProperties" that Gemini doesn't support
步骤4：使用用户查询发送到Gemini（从CLI或提示符）
user_query = "What is the status of vehicle 102?" # example only; client now reads from command line or prompts the user
response = client.models.generate_content(
model="gemini-2.5-flash-lite",
contents=user_query,
config=types.GenerateContentConfig(tools=gemini_tools)
)
# Gemini analyzes: "User wants vehicle 102 info → I should call get_vehicle_details"
# (or "How many are shipped?" → call inventory_summary)
步骤5：处理Gemini的工具请求（可能有多个）
if response.candidates[0].content.parts[0].function_call:
fc = response.candidates[0].content.parts[0].function_call
# fc.name = "get_vehicle_details"
# fc.args = {"vehicle_id": "102"}
print(f"--- Gemini requested tool: {fc.name} with args {fc.args} ---")
步骤6：通过MCP服务器执行工具
result = await session.call_tool(fc.name, fc.args)
# Sends to MCP server:
# → MCP server calls get_vehicle_details("102")
# → MCP server makes HTTP request to data_source
# → Returns formatted result
第7步：将结果发送回Gemini
final_response = client.models.generate_content(
model="gemini-2.5-flash-lite",
contents=[
types.Content(role="user", parts=[types.Part.from_text(text=user_query)]),
response.candidates[0].content, # Original Gemini response
types.Content(
role="tool",
parts=[
types.Part.from_function_response(
name=fc.name,
response={"result": result.content[0].text}
)
]
)
],
config=types.GenerateContentConfig(tools=gemini_tools)
)
# Gemini now has: "User asked → I called tool(s) → Tool returned data"
# Gemini generates final human-readable answer
步骤8：提取并显示答案
answer = final_response.text
print(f"\nFinal AI Answer: {answer}")
# Example output: "Honda Fit is currently shipped heading to Kandy"
关键助手功能：
def clean_schema(schema):
"""Remove additionalProperties from schema as Gemini doesn't support it"""
# Recursively removes "additionalProperties" from JSON schema
# This is needed because MCP tools generate schemas with this field,
# but Google Gemini API doesn't accept it
要运行：
# First, ensure data source is running in another terminal
python data_source.py
# Then in another terminal, run the client, supplying a query as arguments
python mcp_client.py "How many cars are in port?"
# or simply run without args and answer the prompt:
python mcp_client.py
# Enter your query: Find all Toyotas headed to Colombo
# You will see the available tools listed and then
# one or more tool calls. Example:
# --- Gemini requested tool: search_vehicles with args {'status': 'shipped'} ---
# --- Gemini requested tool: inventory_summary with args {} ---
# Final AI Answer: I found that there is 1 vehicle that has been shipped. Would you like to know its details?
🔄 完成消息流
示例：用户询问“车辆102的状态如何？”（或“查找所有前往科伦坡的丰田汽车”）
User
│
└──> mcp_client.py
│
├─ Starts mcp_server.py (as subprocess)
│
├─ Connects to Google Gemini API
│
├─ Gets tool list from MCP Server
│ └─ [get_vehicle_details, search_vehicles, inventory_summary, change_status]
│
├─ Sends to Gemini:
│ "Here are available tools: [get_vehicle_details]
│ User query: What is the status of vehicle 102?"
│
├─ Gemini responds:
│ "I should call get_vehicle_details with vehicle_id=102"
│
├─ mcp_client calls MCP Server:
│ "Execute get_vehicle_details({'vehicle_id': '102'})"
│
├─ mcp_server.py receives call
│ ├─ May call HTTP GET http://127.0.0.1:8001/vehicles/102
│ ├─ Or GET http://127.0.0.1:8001/vehicles/search?status=Shipped
│ └─ Or PATCH to update status
│
├─ data_source.py responds:
│ {"make": "Honda", "model": "Fit", "status": "Shipped", "destination": "Kandy"}
│
├─ mcp_server returns formatted:
│ "Vehicle 102: Honda Fit is currently Shipped heading to Kandy."
│
├─ mcp_client gets result from MCP Server
│
├─ Sends to Gemini:
│ "Tool returned: Vehicle 102: Honda Fit is currently Shipped heading to Kandy."
│
├─ Gemini generates final answer:
│ "The Honda Fit (vehicle 102) is currently shipped and heading to Kandy."
│ (or "One vehicle is shipped; details: Vehicle 102 …")
│
└──> Final Answer displayed to user
🚀 设置和运行
先决条件
pip install fastapi uvicorn google-genai mcp fastmcp httpx python-dotenv
环境变量
创建一个 .env 文件：
GOOGLE_API_KEY=your_actual_api_key_here
运行顺序（重要！）
终端1：启动数据源API
python data_source.py
# Output: INFO: Uvicorn running on http://127.0.0.1:8001
终端2：运行MCP客户端
python mcp_client.py
# Output:
# --- Gemini requested tool: get_vehicle_details with args {'vehicle_id': '102'} ---
# Final AI Answer: ...
🔧 每个组件的工作原理
| 组件 | 角色 | 技术 | 沟通 |
|---|---|---|---|
| data_source.py | 数据提供者 | FastAPI REST | HTTP/JSON |
| mcp_server.py | 工具包装器 | FastMCP | stdio（stdin/stdout） |
| mcp_client.py | 编排器 | MCP客户端+Genai SDK | 协议缓冲区+HTTP |
| API双子星 | 人工智能大脑 | 谷歌 | HTTP REST API |
🐛 故障排除
问题：数据源上的“连接被拒绝”
解决方案：确保 data_source.py 正在端口8001上运行
问题：“超出Gemini配额”
解决方案：使用 gemini-2.5-flash-lite 而不是 gemini-2.0-flash （更好的免费配额）
问题：“x-api-key无效”
解决方案：设置 GOOGLE_API_KEY 具有有效密钥的环境变量
问题：超时错误
解决方案：MCP服务器超时5秒。确保数据源在5秒内响应
📝 关键概念
- MCP（模型上下文协议）：将AI模型与外部工具连接的标准方式
- 工具定义：描述函数输入/输出的模式
- 函数调用：AI模型决定根据用户查询调用工具
- 工具执行：系统执行函数并返回结果
- 能动循环：推理循环→ 工具调用→ 结果→ 最终答案
🎯 定制
添加更多工具
编辑 mcp_server.py:
@mcp.tool()
async def get_vehicle_location(vehicle_id: str) -> str:
"""Get current location of vehicle"""
# Your implementation
pass
更改查询
编辑 mcp_client.py:
# The script now reads the query from command-line args or a prompt.
# You can still modify this default placeholder if you like:
user_query = "What is the location of vehicle 102?" # Change this
切换模型
编辑 mcp_client.py:
model_id = "gemini-2.5-pro" # or any other available model
✅ 下一步
- 探索使用 搜索车辆 用于基于标准的查找
- 请代理人 汇总库存“港口有多少辆车？”
- 使用 change_status 以编程方式修改状态
README现在反映了最新的工具集和代理功能。
📚 参考文献
创建: 2026-02-25
建筑：三倍（AI+协议桥+数据API）
状态：工作✅
