input_analyze
Analyzes text using Baidu Cloud AI Content Safety MCP Server to identify and flag potential security risks or unsafe content for moderation purposes.
Instructions
Name:
检测文本安全
Description:
将文本进行检测，判断是否存在风险内容
Args:
text: 待检测文本
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| text | Yes |
Input Schema (JSON Schema)
{
"properties": {
"text": {
"title": "Text",
"type": "string"
}
},
"required": [
"text"
],
"title": "input_analyzeArguments",
"type": "object"
}
Implementation Reference
- The core handler function for the 'input_analyze' tool. It takes input text, authenticates and calls Baidu's LLM input safety analysis API (/rcs/llm/input/analyze), and returns the analysis result.@mcp.tool() async def input_analyze( text: str, ctx: Context ) -> dict: """ Name: 检测文本安全 Description: 将文本进行检测，判断是否存在风险内容 Args: text: 待检测文本 """ try: credentials = BceCredentials(access_key_id, secret_access_key) # 填写ak、sk # API接口的请求方法 http_method = "POST" # 接口请求路径 input_path = "/rcs/llm/input/analyze" # -----------------------输入安全------------------------------ # 接口请求的header头 headers = { "host": "afd.bj.baidubce.com", "content-type": "application/json; charset=utf-8", "x-bce-date": datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"), } # 设置参与鉴权的时间戳 timestamp = int(time.time()) # 接口请求参数 params = {} # 接口请求的body数据 body = { "query": text, "appid": "", "historyQA": [] } # 设置参与鉴权编码的header，即headers_to_sign,至少包含host，百度智能云API的唯一要求是Host域必须被编码 headers_to_sign = { "host", "x-bce-date", } # 设置到期时间，默认1800s expiration_in_seconds = 18000 # 生成鉴权字符串 result = sign(credentials, http_method, input_path, headers, params, timestamp, expiration_in_seconds, headers_to_sign) headers['authorization'] = result # 拼接接口的url地址 url = 'http://%s%s' % (headers['host'], input_path) async with httpx.AsyncClient() as client: response = await client.post(url, headers=headers, json=body) response.raise_for_status() result = response.json() # if result.get("status") != 0: # error_msg = result.get("message", "unkown error") # raise Exception(f"API response error: {error_msg}") result['req'] = body return result except httpx.HTTPError as e: raise Exception(f"HTTP request failed: {str(e)}") from e except KeyError as e: raise Exception(f"Failed to parse reponse: {str(e)}") from e