AIGC入门,程序员的第一个MCP例子
本文展示了一个基于MCP协议的时间查询服务器示例。该Python脚本提供了四个主要功能:1)获取指定时区的当前时间并可自定义格式;2)查询特定时区的详细信息;3)列出常用时区可按地区过滤;4)计算两个时区之间的时间差。服务器通过标准输入/输出(stdio)方式运行,使用pytz库处理时区转换,并提供了清晰的错误处理机制。每个功能都定义了输入参数模式,支持默认值设置,确保接口的易用性。
一步步来,先来stdio的Claude MCP例子
MCP示例:时间查询
创建目录及mcp示例脚本
/Users/mac/mcp_time_server/time_server.py
#!/usr/bin/env python3
"""
时间查询MCP服务器
支持多时区时间查询和格式化
"""
import asyncio
import logging
from datetime import datetime
import pytz
from typing import Any, Sequence
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("time-server")
# 创建服务器实例
server = Server("time-query-server")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
返回可用工具列表
"""
return [
types.Tool(
name="get_current_time",
description="获取当前时间,支持指定时区",
inputSchema={
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区名称 (如: Asia/Shanghai, UTC, America/New_York)",
"default": "UTC"
},
"format": {
"type": "string",
"description": "时间格式 (如: %Y-%m-%d %H:%M:%S)",
"default": "%Y-%m-%d %H:%M:%S"
}
}
}
),
types.Tool(
name="get_timezone_info",
description="获取时区信息和当前时间",
inputSchema={
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区名称 (如: Asia/Shanghai, UTC, America/New_York)",
"default": "UTC"
}
}
}
),
types.Tool(
name="list_timezones",
description="列出常用时区",
inputSchema={
"type": "object",
"properties": {
"region": {
"type": "string",
"description": "地区过滤 (如: Asia, America, Europe)",
"default": ""
}
}
}
),
types.Tool(
name="time_difference",
description="计算两个时区之间的时间差",
inputSchema={
"type": "object",
"properties": {
"timezone1": {
"type": "string",
"description": "第一个时区",
"default": "UTC"
},
"timezone2": {
"type": "string",
"description": "第二个时区",
"default": "Asia/Shanghai"
}
}
}
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent]:
"""
处理工具调用
"""
if arguments is None:
arguments = {}
try:
if name == "get_current_time":
timezone_name = arguments.get("timezone", "UTC")
time_format = arguments.get("format", "%Y-%m-%d %H:%M:%S")
try:
tz = pytz.timezone(timezone_name)
current_time = datetime.now(tz)
formatted_time = current_time.strftime(time_format)
return [
types.TextContent(
type="text",
text=f"当前时间 ({timezone_name}): {formatted_time}"
)
]
except Exception as e:
return [
types.TextContent(
type="text",
text=f"错误: 无效的时区或格式 - {str(e)}"
)
]
elif name == "get_timezone_info":
timezone_name = arguments.get("timezone", "UTC")
try:
tz = pytz.timezone(timezone_name)
current_time = datetime.now(tz)
utc_offset = current_time.strftime("%z")
info = f"""时区信息:
- 时区名称: {timezone_name}
- 当前时间: {current_time.strftime("%Y-%m-%d %H:%M:%S")}
- UTC偏移: {utc_offset}
- 时区缩写: {current_time.strftime("%Z")}"""
return [types.TextContent(type="text", text=info)]
except Exception as e:
return [
types.TextContent(
type="text",
text=f"错误: 无效的时区 - {str(e)}"
)
]
elif name == "list_timezones":
region = arguments.get("region", "").strip()
# 常用时区列表
common_timezones = [
"UTC", "GMT",
"Asia/Shanghai", "Asia/Tokyo", "Asia/Seoul", "Asia/Hong_Kong",
"America/New_York", "America/Los_Angeles", "America/Chicago",
"Europe/London", "Europe/Paris", "Europe/Berlin", "Europe/Rome",
"Australia/Sydney", "Australia/Melbourne"
]
if region:
filtered_timezones = [tz for tz in common_timezones if region in tz]
else:
filtered_timezones = common_timezones
timezone_list = "".join([f"- {tz}" for tz in filtered_timezones])
return [
types.TextContent(
type="text",
text=f"{'过滤后的' if region else '常用'}时区列表:{timezone_list}"
)
]
elif name == "time_difference":
tz1_name = arguments.get("timezone1", "UTC")
tz2_name = arguments.get("timezone2", "Asia/Shanghai")
try:
tz1 = pytz.timezone(tz1_name)
tz2 = pytz.timezone(tz2_name)
now = datetime.now(pytz.UTC)
time1 = now.astimezone(tz1)
time2 = now.astimezone(tz2)
# 计算时差
offset1 = time1.utcoffset().total_seconds() / 3600
offset2 = time2.utcoffset().total_seconds() / 3600
diff = offset2 - offset1
result = f"""时区时间对比:
- {tz1_name}: {time1.strftime("%Y-%m-%d %H:%M:%S")} (UTC{offset1:+.0f})
- {tz2_name}: {time2.strftime("%Y-%m-%d %H:%M:%S")} (UTC{offset2:+.0f})
- 时差: {diff:+.0f} 小时"""
return [types.TextContent(type="text", text=result)]
except Exception as e:
return [
types.TextContent(
type="text",
text=f"错误: 无效的时区 - {str(e)}"
)
]
else:
return [
types.TextContent(
type="text",
text=f"错误: 未知工具 '{name}'"
)
]
except Exception as e:
return [
types.TextContent(
type="text",
text=f"执行工具 '{name}' 时出错: {str(e)}"
)
]
async def main():
"""主函数"""
logger.info("启动时间查询MCP服务器...")
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="time-query-server",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
)
)
if __name__ == "__main__":
asyncio.run(main())
进行Claude配置
/Users/mac/Library/Application Support/Claude/claude_desktop_config.json
{
"mcpServers": {
"time-query-server": {
"command": "/Users/mac/my_venv/bin/python",
"args": ["/Users/mac/mcp_time_server/time_server.py"],
"env": {
"VIRTUAL_ENV": "/Users/mac/my_venv"
},
"description": "时间查询MCP服务器 - 支持多时区时间查询和格式化",
"disabled": false
}
}
}
重启Claude
看到下方2个截图则表示安装正确
测试MCP-查询当前时间
可以看到Claude自动使用了这个MCP服务了
MCP工具选择推理过程
我有多个可用的工具,其中time-query-server明显最适合:
● 工具描述:“获取当前时间,支持指定时区”
● 其他工具如baidu-search、web_search等都不是专门处理时间查询的
参数构造推理
-
时区参数推理:
○ 大阪在日本
○ 日本标准时区是 Asia/Tokyo(东京时区覆盖整个日本,包括大阪)
○ 这是标准的IANA时区标识符格式 -
格式参数选择:
○ 使用默认的 %Y-%m-%d %H:%M:%S 格式
○ 这是最常见和清晰的时间显示格式
知识来源
这些推理基于我的训练数据中包含的:
● 世界地理知识(大阪在日本)
● 时区知识(日本使用Asia/Tokyo时区)
● 工具使用模式和API设计常识
● IANA时区数据库的标准命名规则
所以整个过程是:意图识别 → 工具匹配 → 参数映射 → API调用。
Claude MCP小结
● 脚本编写可以给出提示词Claude帮你生成
● MCP服务脚本编写完成后,并不需要在item2中通过命令行主动运行起来,即不需要执行python3 time_server.py,下方是Claude启动前后的进程对比,可以看到Claude启动后自动运行了time_server.py脚本了
一步步来,再来个stdio协议的IntelliJ IDEA MCP例子
MCP示例:git commit时,message自动在结尾增加一行文字
创建目录及mcp示例脚本
/Users/mac/.aone_copilot/mcp_servers/commit-message-enhancer/main.py
#!/usr/bin/env python3
import json
import sys
import asyncio
# 模拟FastMCP类,避免依赖问题
class FastMCP:
def __init__(self, name: str):
self.name = name
self.tools = {}
def tool(self, description: str):
def decorator(func):
self.tools[func.__name__] = {
"function": func,
"description": description
}
return func
return decorator
def run(self):
# 简单的MCP协议实现
if len(sys.argv) > 1 and sys.argv[1] == "--mcp":
if len(sys.argv) > 2 and sys.argv[2] == "tools":
# 返回工具列表
tools_list = []
for name, tool in self.tools.items():
tools_list.append({
"name": name,
"description": tool["description"],
"inputSchema": {
"type": "object",
"properties": {},
"required": []
}
})
print(json.dumps(tools_list))
elif len(sys.argv) > 2 and sys.argv[2] == "tool" and len(sys.argv) > 3:
# 执行工具
tool_name = sys.argv[3]
if tool_name in self.tools:
# 获取函数参数(如果有)
kwargs = {}
if len(sys.argv) > 4:
try:
kwargs = json.loads(sys.argv[4])
except:
pass
# 执行函数
result = asyncio.run(self.tools[tool_name]["function"](**kwargs))
output = {
"content": [{
"type": "text",
"text": result
}]
}
print(json.dumps(output))
# 创建 MCP 服务器
mcp = FastMCP("Commit Message Enhancer")
# MCP 工具 - 增强commit message
@mcp.tool(description="在commit message末尾添加协作开发信息")
async def enhance_commit_message(message: str) -> str:
"""
在commit message末尾添加协作开发信息
Args:
message: 原始commit message
"""
try:
# 在commit message末尾添加协作开发信息
enhanced_message = message + "\n\nCo-developed-by: Lingma Copilot <lingma-agent@alibaba-inc.com>"
return enhanced_message
except Exception as e:
return f"增强commit message时出错: {str(e)}"
if __name__ == "__main__":
mcp.run()
配置后原则上无须重启IDEA即可加载出来,如下图所示
测试MCP-提交代码
Aone Copilot插件-Agent模式下,输入“提交代码”,会自动完成代码提交工作
总结
学会查看错误日志
如果配置后Claude Destop重启后依旧看不到或者有报错信息,可以使用下面的命令查看错误详情tail -n 20 -F ~/Library/Logs/Claude/mcp*.log
更多推荐
所有评论(0)