蓝心九问
工具库插件工具配置MCP

IDE MCP

mcp工具

工具申明

使用 @mcp.tool() 装饰器将python方法申明为mcp工具。

from pydantic import Field

@mcp.tool(description="获取指定城市的天气信息")
async def get_weather(city: str = Field(description="需要查询天气的城市"),
                date: str = Field(description="查询天气的日期")) -> dict:
    return {}

tips：

  1. 不用关心FastMCP的创建和main方法，蓝心九问会动态添加到代码中，fastmcp版本2.11.0，mcp版本1.11.0。

  2. 使用@mcp.tool()装饰器声明工具，工具可以申明多个但不要重名。

  3. 使用description字段设置工具描述，如果为空则会将方法注释作为工具描述。

  4. 工具执行超时时间180秒。

入参申明

mcp工具入参支持如下类型：

类型说明
int整数类型
float浮点数类型
bool布尔值类型
str字符串类型
dict字典类型
enum枚举类型
pydantic model自定义对象类型
list[int]整数列表类型
list[float]浮点数列表类型
list[bool]布尔值列表类型
list[str]字符串列表类型
list[dict]字典列表类型
list[pydantic model]自定义对象列表类型

简单类型

简单类型指的是int、float、str、bool、dict以及他们的数组类型。

from pydantic import Field

@mcp.tool(description="获取指定城市的天气信息")
async def get_weather(date: str = Field(default="2025-10-10", description="查询天气的日期")) -> dict:
    return {}

使用pydantic Field类辅助申明工具入参。

  1. default：定义该参数默认值。

  2. description：参数描述，用于告知大语言模型该参数的含义。

default说明

mcp sdk会使用pydantic validator对方法入参进行校验，pydantic validator要求没有默认值的参数放前面，有默认值的参数放后面，如下：

def fn(没有默认值参数,有默认值参数)

这种校验和python语法是保持一致的，与python语法一样，有下面两种方法通过校验：

  1. 遵循校验规则，有默认值参数全部放后面。

  2. 方法入参前面加上 * ,如下：

from pydantic import Field

@mcp.tool(description="获取指定城市的天气信息")
async def get_weather(
        *,
        date: str = Field(default="2025-10-10", description="查询天气的日期"),
        city: str = Field(description="城市")
) -> dict:
    return {}

自定义类型

自定义类型指的是python自定义类，该类需要继承pydantic BaseModel。

1.定义类型

from pydantic import Field, BaseModel

class User(BaseModel):
    user_name: str = Field(default="哈哈", description="用户信息")
    age: int = Field(description="年龄")

2.使用类型

@mcp.tool(description="保存用户信息")
async def save_user(user: User = Field(description="用户信息")) -> dict:
    return {}

@mcp.tool(description="批量保存用户信息")
async def save_users(user: list[User] = Field(description="用户信息列表")) -> dict:
    return {}

3.支持嵌套

自定类型支持再嵌套自定义类型，但是不支持自定义类型再嵌套自己，递归解析出现栈溢出，如下面User类中存在User类型的属性

class User(BaseModel):
    user_name: str = Field(description="用户信息")
    age: int = Field(description="年龄")
    user: User = Field(description="这个属性又是User类型，产生循环嵌套，不支持")

如何获取系统参数

系统参数指的是蓝心九问平台参数，例如用户工号、登录态信息等，详情如下:

系统参数类型说明
sys_openidstr工号，只有需要登录的场景才有
sys_uuc_infodict登录态信息, 包含token、uucToken、uuid
sys_from_debugbool是否来自调试

蓝心九问系统参数采用申明注入的方式(只要申明的参数名与系统参数名一致就会自动注入改参数)，如下：

@mcp.tool(description="获取指定城市，指定时间的天气信息")
async def get_weather(
        sys_openid:str,
        sys_uuc_info:dict,
        sys_from_debug:bool,
) -> dict:

    return {}

工具返回

mcp工具方法最好申明返回值类型，建议返回值类型为为dict或者pydantic BaseModel。

from pydantic import Field, BaseModel

@mcp.tool(description="保存用户信息")
async def save_user(user: User = Field(description="用户信息")) -> dict:
    return {}

class Result(BaseModel):
    user_id:str=Field(description="用户id")

@mcp.tool(description="保存用户信息")
async def save_user(user: User = Field(description="用户信息")) -> Result:
    return Result(user_id=user.id)

输出schema生成

mcp在调试时会生成mcp工具输出schema，这个schema主要用于工作流插件节点的输出定义，如下图：

image.png

基于sdk生成(☆推荐)

当工具返回为pydantic BaseModel时(方法申明)，mcp sdk会基于 BaseModel的定义生成对应的outputSchema。

基于输出逆向生成

如果mcp工具返回的参数类型为dict(方法申明)时，mcp会根据调试结果逆向生成outputSchema 例如某mcp工具输出结果如下：

{
  "news": [
    {
      "id": 0,
      "title": "世预赛最后一战！国足vs巴林",
      "desc": "",
      "cover": "https://fyb-1.cdn.bcebos.com/fyb/de6163834f53ca92c1273fff98ac9078.jpeg",
      "author": "",
      "timestamp": 0,
      "hot": 7808565,
      "url": "https://www.baidu.com/s?wd=hello",
      "mobileUrl": "https://m.baidu.com/s?word=hello"
    }
  ]
}

基于这个输出结果逆向生成出来的outputSchema是和下面的pydantic BaseModel是等价的。

from pydantic import Field, BaseModel

class New(BaseModel):
    id:int=Field(description="")
    title:int=Field(description="")
    ……

class Result(BaseModel):
    news:list[New]=Field(description="")

基于逆向生成出来的outputSchema缺少描述信息，而且要想生成outputSchema完整，需要工具返回的dict包含全部的key，并且value不能为None，list不能没有元素。

手动设置

如果mcp方法返回为dict时，针对上面如果返回的dict少key、有None值或者空list，导致逆向生成出来的outputSchema不完整

九问mcp工具提供了一种自定义outputSchema方式(此种方式的优先级最高)，具体使用方式如下图所示：

image.png

在返回dict中增加$$mcp_output_schema字段，其值为一个自定义dict，当检测到返回的dict包含该字段时，会根据该字段的值生成schema，达到自定义schema的目的，这是一种兜底方案，建议使用默认生成

代码规范

蓝心九问调试mcp时会对代码进行校验，不遵循下面的规范会导致校验不通过。

禁止使用print函数，在mcp中禁止向标准输出流(stdout)中写入数据，不然会污染mcp的协议通道，这一点很重要。

不要导入使用不到的包，例如下面代码中的time。

import time

@mcp.tool(description="测试")
async def test() -> dict:
    return {}

自定义类之间依赖，以类B依赖类A为例，下面两点遵循一个即可(python语言要求)。

①被依赖的类写在上面，类A声明在B的前面。

from pydantic import BaseModel, Field

class A(BaseModel):
    name: str = Field(description="")


class B(BaseModel):
    a: A = Field(description="")

②被依赖的类作为属性时加上单引号，例如下面类B中a属性。

from pydantic import BaseModel, Field

class B(BaseModel):
    a: 'A' = Field(description="")


class A(BaseModel):
    name: str = Field(description="")

禁用库

目前蓝心九问mcp并没有实现运行环境隔离，为了保障mcp服务运行安全与稳定，蓝心九问禁用了一些python库，如下：

[
    "os",
    "builtins",
    "threading",
    "resource",
    "shutil",
    "pathlib",
    "tempfile",
    "subprocess",
    "multiprocessing",
    "requests",
    "httpx",
    "socket",
    "http",
    "aiohttp",
    "logging",
    "sys",
    "ctypes",
    "mmap",
    "importlib",
    "imp",
    "runpy",
    "pkgutil",
    "pickle",
    "cPickle",
    "marshal",
    "shelve",
    "pty",
    "tty",
    "termios",
    "shlex",
    "fcntl",
    "pwd",
    "grp",
    "spwd",
    "signal",
    "sysconfig",
    "platform",
    "urllib",
    "ssl"
]

内置库

下面是蓝心九问官方提供的SDK。

📢 重要重要: 建议使用异步SDK，同步SDK会导致整个MCP服务的并发能力比较弱。

async_http_client(☆推荐)

async_http_client用于异步请求第三方服务

async_http_client基于httpx封装，使用方式与requests类似，支持GET、PUT、POST、DELETE。

默认timeout=3，需要是数值类型，不支持元组类型。

from pydantic import Field
import async_http_client

@mcp.tool(description="测试")
async def test() -> dict:
    response = await async_http_client.get(url="http://127.0.0.1:5001/health")
    return {"response":response.json()}

http_client

async_http_client的同步版本

http_client基于requests封装，使用方式与requests类似，支持GET、PUT、POST、DELETE。

默认timeout=3，需要是数值类型，不支持元组类型。

import http_client

@mcp.tool(description="测试")
async def test() -> dict:
    response = http_client.get(url="http://127.0.0.1:5001/health")
    return {"response":response.json()}

async_http_proxy_client(☆推荐)

async_http_proxy_client用于线上环境异步请求蛇口机房(测试环境)部署的服务

async_http_proxy_client只能用于简单的接口转发，其能力相较于requests包有较大功能缺失，支持GET、POST。

默认timeout=3。

import async_http_proxy_client

@mcp.tool(description="测试")
async def test() -> dict:
    params = {"param": "params"}
    cookies = {"cookie":"cookie"}
    headers = {"header":"header"}

    # for get
    response_get = await async_http_proxy_client.get(
        url="", 
        params=params, 
        cookies=cookies,
        headers=headers,
        timeout=2
    )

    # for post
    payload = {"payload": "payload"}
    response_post = await async_http_proxy_client.post(
        url="", 
        params=params, 
        json=payload, 
        cookies=cookies,
        headers=headers,
        timeout=2
    )

    if response_get.status_code != 200:
        raise RuntimeError("请求失败")

    return response_get.json()

http_proxy_client

async_http_proxy_client的同步版本

async_bot_caller(☆推荐)

异步调用蓝心九问Bot

from async_bot_caller import AsyncBotCaller

@mcp.tool()
async def test()->dict:
    async_bot_caller = AsyncBotCaller(
        bot_id=1204,
        authorization="app-hQlVZw3weqIGT8tyL1oj904R",
        env="test"
    )
    result = await async_bot_caller.call(
        query="你好啊",
        input={}, # 预置变量
        timeout=10
    )

env指的是当前使用蓝心九问的环境(测试:test、预发:pre、线上:prd),不是跨环境调用(如:线上调用测试环境插件)

bot_caller

async_bot_caller的同步版本

from bot_caller import BotCaller

@mcp.tool()
async def test()->dict:
    bot_caller = BotCaller(
        bot_id=1204,
        authorization="app-hQlVZw3weqIGT8tyL1oj904R",
        env="test"
    )
    result = bot_caller.call(
        query="你好啊",
        input={}, # 预置变量
        timeout=10
    )

env指的是当前使用蓝心九问的环境(测试:test、预发:pre、线上:prd),不是跨环境调用(如:线上调用测试环境插件)

async_plugin_caller(☆推荐)

异步调用其他已发布的工具或者mcp

from async_plugin_caller import AsyncPluginCaller

@mcp.tool()
async def test()->dict:
    async_plugin_caller = AsyncPluginCaller(plugin_id=967, env="test")
    result = await async_plugin_caller.call(
        tool_name = "query_site",
        arguments = {"siteId": "wk25070212d406c6"},
        timeout=10
    )

env指的是当前使用蓝心九问的环境(测试:test、预发:pre、线上:prd),不是跨环境调用(如:线上调用测试环境插件)

plugin_caller

async_plugin_caller的异步版

from plugin_caller import PluginCaller

@mcp.tool()
async def test()->dict:
    plugin_caller = PluginCaller(plugin_id=967, env="test")
    result = plugin_caller.call(
        tool_name = "query_site",
        arguments = {"siteId": "wk25070212d406c6"},
        timeout=10
    )

env指的是当前使用蓝心九问的环境(测试：test、预发：pre、线上：prd)，不是跨环境调用(如线上调用测试环境插件)

HTTP

Previous Page

远程MCP

Next Page

On this page

mcp工具
工具申明
入参申明
简单类型
default说明
自定义类型
如何获取系统参数
工具返回
输出schema生成
基于sdk生成(☆推荐)
基于输出逆向生成
手动设置
代码规范
禁用库
内置库
async_http_client(☆推荐)
http_client
async_http_proxy_client(☆推荐)
http_proxy_client
async_bot_caller(☆推荐)
bot_caller
async_plugin_caller(☆推荐)
plugin_caller