跳转到主要内容
MCP(Model Context Protocol)工具是使用 @mcp.tool() 装饰器的函数,用于扩展 AI 智能体的能力。

基本示例

文件:mcp/tools/math.py
from server import mcp

@mcp.tool()
def multiply(first_val: float, second_val: float) -> float:
    '''
    Calculate the product of two numbers

    Args:
        first_val: the first value to be multiplied
        second_val: the second value to be multiplied
    '''
    return round(first_val * second_val, 4)

关键要求

1. 使用 @mcp.tool() 装饰器

所有工具必须使用 @mcp.tool() 装饰器,才能被 Osmosis 发现。

2. 放置在 mcp/tools/ 目录中

将你的工具模块组织在 mcp/tools/ 目录中。

3. 包含类型注解

所有参数和返回值都需要类型注解:
def my_tool(input_text: str, count: int) -> list[str]:
    # Implementation
    pass

4. 添加文档字符串

详尽的文档字符串有助于 AI 理解何时以及如何使用你的工具:
@mcp.tool()
def search_database(query: str, limit: int = 10) -> list[dict]:
    """
    Search the database for matching records

    Args:
        query: Search query string
        limit: Maximum number of results to return (default: 10)

    Returns:
        List of matching records as dictionaries
    """
    # Implementation
    pass

5. 在 __init__.py 中导出

mcp/tools/__init__.py 中导出你的工具:
from .math import multiply
from .search import search_database

__all__ = ['multiply', 'search_database']

服务器配置

FastMCP 服务器配置

文件:mcp/server/mcp_server.py
from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse

mcp = FastMCP("OsmosisTools")

@mcp.custom_route("/health", methods=["GET"])
async def health(request: Request) -> JSONResponse:
    return JSONResponse({"status": "healthy"})

入口文件

文件:mcp/main.py
import argparse
from server.mcp_server import mcp
from tools import *  # Import all tools

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="0.0.0.0")
    parser.add_argument("--port", type=int, default=8080)
    args = parser.parse_args()

    mcp.run(transport="http", host=args.host, port=args.port)

完整示例

以下是一个更复杂的工具示例,展示了最佳实践:
from server import mcp
from typing import Optional
import json

@mcp.tool()
def fetch_user_profile(
    user_id: str,
    include_history: bool = False,
    max_items: Optional[int] = None
) -> dict:
    """
    Fetch a user's profile information from the database

    This tool retrieves user profile data and optionally includes
    their activity history. Use this when you need to look up
    information about a specific user.

    Args:
        user_id: Unique identifier for the user
        include_history: Whether to include activity history (default: False)
        max_items: Maximum number of history items to return (optional)

    Returns:
        Dictionary containing user profile data with keys:
        - id: User ID
        - name: User's full name
        - email: User's email address
        - history: Activity history (if include_history=True)

    Raises:
        ValueError: If user_id is not found
    """
    # Your implementation here
    profile = {
        "id": user_id,
        "name": "John Doe",
        "email": "john@example.com"
    }

    if include_history:
        history = get_user_history(user_id, max_items)
        profile["history"] = history

    return profile

测试你的工具

本地测试

在本地启动你的 MCP 服务器:
python mcp/main.py
在另一个终端中测试:
python mcp/test/test.py

单元测试

为你的工具创建测试:
# tests/test_math.py
from mcp.tools.math import multiply

def test_multiply():
    assert multiply(2.0, 3.0) == 6.0
    assert multiply(0.5, 0.5) == 0.25
    assert multiply(-2.0, 3.0) == -6.0
有关编码规范、测试策略和 CI/CD 设置,请参阅 最佳实践

下一步