Baidu Cloud AI Content Safety MCP Server

Instructions

Name:
    检测文本安全

Description:
    将文本进行检测，判断是否存在风险内容

Args:
    text:  待检测文本

Implementation Reference

• src/mcp_server_safe_content_check/server.py:177-250 (handler)

  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