Python MCP Server 开发指南：从零构建自定义工具服务

1
from fastmcp import FastMCP
2

3
# 创建 MCP Server 实例
4
mcp = FastMCP("my-tools")
5

6

7
@mcp.tool()
8
def get_system_info() -> str:
9
    """返回系统基本信息。
10

11
    返回当前操作系统的名称、版本以及 Python 运行时版本。
12
    """
13
    import platform
14
    return (
15
        f"OS: {platform.system()} {platform.release()}\n"
16
        f"Python: {platform.python_version()}\n"
17
        f"架构: {platform.machine()}"
18
    )
19

20

21
@mcp.tool()
22
def count_files(directory: str, extension: str = "") -> str:
23
    """统计指定目录下的文件数量。
24

25
    Args:
26
        directory: 要扫描的目录路径
27
        extension: 可选，按文件扩展名过滤（如 .py、.json）
28
    """
29
    from pathlib import Path
30

31
    path = Path(directory).expanduser()
32
    if not path.is_dir():
33
        return f"错误：{directory} 不是有效目录"
34

35
    if extension:
36
        files = list(path.rglob(f"*{extension}"))
37
    else:
38
        files = [f for f in path.rglob("*") if f.is_file()]
39

40
    return f"在 {directory} 中找到 {len(files)} 个文件" + (
41
        f"（扩展名 {extension}）" if extension else ""
42
    )
43

44

45
if __name__ == "__main__":
46
    mcp.run(transport="stdio")

1
import httpx
2
from fastmcp import FastMCP
3

4
mcp = FastMCP("my-tools")
5

6

7
@mcp.tool()
8
async def check_website(url: str) -> str:
9
    """检查网站可访问性并返回状态信息。
10

11
    Args:
12
        url: 完整的网站地址（必须包含 https:// 或 http://）
13
    """
14
    async with httpx.AsyncClient(timeout=10.0) as client:
15
        try:
16
            response = await client.get(url)
17
            return (
18
                f"状态码: {response.status_code}\n"
19
                f"响应时间: {response.elapsed.total_seconds():.2f}秒\n"
20
                f"Content-Type: {response.headers.get('content-type', '未知')}"
21
            )
22
        except httpx.RequestError as e:
23
            return f"请求失败 {url}：{str(e)}"
24

25

26
@mcp.tool()
27
async def translate_text(text: str, target_lang: str = "en") -> str:
28
    """调用翻译 API 将文本翻译为目标语言。
29

30
    Args:
31
        text: 要翻译的文本
32
        target_lang: 目标语言代码（如 en、ja、zh）
33
    """
34
    async with httpx.AsyncClient(timeout=15.0) as client:
35
        try:
36
            # 此处使用免费翻译接口作为示例
37
            response = await client.post(
38
                "https://api.mymemory.translated.net/get",
39
                params={"q": text, "langpair": f"|{target_lang}"},
40
            )
41
            data = response.json()
42
            if data.get("responseStatus") == 200:
43
                return data["responseData"]["translatedText"]
44
            return f"翻译失败：{data.get('responseDetails', '未知错误')}"
45
        except Exception as e:
46
            return f"翻译请求异常：{str(e)}"
47

48

49
if __name__ == "__main__":
50
    mcp.run(transport="stdio")

1
{
2
  "mcpServers": {
3
    "my-tools": {
4
      "url": "http://localhost:8080/sse"
5
    }
6
  }
7
}

1
{
2
  "mcpServers": {
3
    "my-tools": {
4
      "command": "/path/to/my-mcp-server/.venv/bin/python",
5
      "args": ["/path/to/my-mcp-server/server.py"]
6
    }
7
  }
8
}

1
{
2
  "mcpServers": {
3
    "my-dify-app": {
4
      "url": "https://api.dify.ai/mcp/srv-xxxxxxxxxxxx"
5
    }
6
  }
7
}