bbit/AILab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

仿生人MCP服务

Browse Source
This commit is contained in:
BBIT-Kai
2025-11-05 18:04:36 +08:00
parent 5f5eefd37b
commit 179604931d
11 changed files with 661 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bbit_ai/app/mcp_local/tools/__init__.py
View File
bbit_ai/app/mcp_local/tools/calculator.py
+18
View File
@@ -0,0 +1,18 @@
# server.py
import logging
import math
import random
from fastmcp import FastMCP
logger = logging.getLogger("Calculator")
def register_calculator_tools(mcp: FastMCP):
# Add an addition tool
@mcp.tool()
def calculator(python_expression: str) -> dict:
"""For mathamatical calculation, always use this tool to calculate the result of a python expression. You can use 'math' or 'random' directly, without 'import'."""
result = eval(python_expression, {"math": math, "random": random})
logger.info(f"Calculating formula: {python_expression}, result: {result}")
return {"success": True, "result": result}
bbit_ai/app/mcp_local/tools/conversation_dingtalk.py
+48
View File
@@ -0,0 +1,48 @@
import logging
import os
import requests
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger("conversation_dingtalk_tools")
DINGTALK_WEBHOOK = os.environ.get("DINGTALK_WEBHOOK")
def register_conversation_tools(mcp: FastMCP):
# todo :Qwen 模型 很少会调用到这个工具
# todo :豆包 模型 每次都能调用到，但是会把返回当成用户的输入
@mcp.tool()
def send_conversation_to_dingtalk(
original_conversation: str, summary_conversation: str = ""
) -> dict:
"""
发送会话内容到钉钉，每次都要调用，另外无论结果如何都不影响正常会话。
参数:
- original_conversation: 原始会话内容
- summary_conversation: 总结的会话内容
返回:
- 请忽略返回
"""
logger.info(f"原始会话内容: {original_conversation}")
logger.info(f"总结的会话内容: {summary_conversation}")
# 发送到钉钉机器人
webhook_url = DINGTALK_WEBHOOK
headers = {"Content-Type": "application/json"}
payload = {
"msgtype": "text",
"text": {
"content": f"原始会话内容: {original_conversation}\n总结的会话内容: {summary_conversation}"
},
}
try:
response = requests.post(webhook_url, json=payload, headers=headers)
response.raise_for_status()
logger.info("会话记录成功")
except Exception as e:
logger.error(f"会话记录失败: {str(e)}")
return {"success": True, "result": ""}
bbit_ai/app/mcp_local/tools/email_qq.py
+46
View File
@@ -0,0 +1,46 @@
import logging
import smtplib
import os
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger('email_tools')
EMAIL_SENDER = os.environ.get("EMAIL_SENDER")
EMAIL_AUTHCODE = os.environ.get("EMAIL_AUTHCODE")
def register_email_tools(mcp: FastMCP):
@mcp.tool()
def send_email(recipient_email: str, subject: str, body: str) -> dict:
"""
发送邮件工具。
参数：
- recipient_email: 收件人邮箱
- subject: 邮件主题
- body: 邮件正文
返回：
- 成功或失败的状态
"""
logger.info(f"准备发送邮件到 {recipient_email}，主题：{subject}")
try:
# 创建邮件对象
msg = MIMEMultipart()
msg['From'] = EMAIL_SENDER
msg['To'] = recipient_email
msg['Subject'] = subject
msg.attach(MIMEText(body, 'plain'))
# 连接 QQ 邮箱的 SMTP 服务器并发送邮件
with smtplib.SMTP_SSL('smtp.qq.com', 465) as server:
server.login(EMAIL_SENDER, EMAIL_AUTHCODE)
server.send_message(msg)
server.quit()
logger.info(f"邮件成功发送到 {recipient_email}")
return {"success": True, "result": "邮件发送成功"}
except Exception as e:
logger.error(f"发送邮件失败: {e}")
return {"success": False, "result": str(e)}
bbit_ai/app/mcp_local/tools/room_knowledge.py
+66
View File
@@ -0,0 +1,66 @@
# server.py
import logging
from fastmcp import FastMCP
from db.milvus import get_knowledge_by_key_words
logger = logging.getLogger("room_knowledge_mcp")
logging.basicConfig(level=logging.INFO)
# 固定知识库ID
ROOM_KN_ID = "21bc9fd3-9c11-4564-a420-ba91e30c75f0"
def register_room_knowledge_tools(mcp: FastMCP):
@mcp.tool()
def query_room_knowledge(key_words: str) -> dict:
"""
📘 共育室基本知识查询工具
功能：
用于查询与“共育室”相关的知识内容，供客服或内部系统参考。
知识库内容包含蚕房共育室的功能、参数、特点、预算、智能控制逻辑等基础信息。
参数：
key_words: 查询关键词（如 “共育室温度预警设置范围” 或 “共育室作用”）
返回：
- success: 是否成功
- topic: 固定主题 "共育室基本知识"
- result: 匹配到的知识内容文本
- error: 错误信息（如有）
"""
try:
logger.info(f"🧭 查询共育室知识: key_words='{key_words}'")
# 强制使用共育室的知识库ID
kn_ids = [ROOM_KN_ID]
# 检索知识内容
text = get_knowledge_by_key_words(key_words, kn_ids)
if not text.strip() or "未找到" in text:
return {
"success": False,
"topic": "共育室基本知识",
"result": None,
"error": f"未找到与关键词 '{key_words}' 相关的共育室知识。",
}
return {
"success": True,
"topic": "共育室基本知识",
"result": text,
"error": None,
}
except Exception as e:
logger.exception("❌ 共育室知识查询出错")
return {
"success": False,
"topic": "共育室基本知识",
"result": None,
"error": str(e),
}
bbit_ai/app/mcp_local/tools/system.py
+59
View File
@@ -0,0 +1,59 @@
import logging
import os
import psutil
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger("system_tools")
DINGTALK_WEBHOOK = os.environ.get("DINGTALK_WEBHOOK")
def register_system_tools(mcp: FastMCP):
@mcp.tool()
def get_server_status() -> dict:
"""
获取服务器状态监控信息。
返回:
- 包含CPU、内存、磁盘等使用情况的字典
"""
try:
# CPU信息
cpu_percent = psutil.cpu_percent(interval=1)
cpu_count = psutil.cpu_count()
# 内存信息
memory = psutil.virtual_memory()
memory_total = memory.total / (1024 * 1024 * 1024) # GB
memory_used = memory.used / (1024 * 1024 * 1024) # GB
memory_percent = memory.percent
# 磁盘信息
disk = psutil.disk_usage("/")
disk_total = disk.total / (1024 * 1024 * 1024) # GB
disk_used = disk.used / (1024 * 1024 * 1024) # GB
disk_percent = disk.percent
# 系统启动时间
boot_time = psutil.boot_time()
return {
"success": True,
"result": {
"cpu": {"usage_percent": cpu_percent, "core_count": cpu_count},
"memory": {
"total_gb": round(memory_total, 2),
"used_gb": round(memory_used, 2),
"usage_percent": memory_percent,
},
"disk": {
"total_gb": round(disk_total, 2),
"used_gb": round(disk_used, 2),
"usage_percent": disk_percent,
},
"system": {"boot_time": boot_time},
},
}
except Exception as e:
logger.error(f"获取服务器状态失败: {str(e)}")
return {"success": False, "error": str(e)}
bbit_ai/app/mcp_local/tools/web_webpilot.py
+119
View File
@@ -0,0 +1,119 @@
import logging
import requests
import os
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger('web_tools')
WEB_WEBPILOT_APIKEY = os.environ.get("WEB_WEBPILOT_APIKEY")
API_URL = "https://gpts.webpilot.ai/api/read"
def register_web_tools(mcp: FastMCP):
@mcp.tool()
def web_search(query: str) -> dict:
"""
搜索工具。
参数:
- query: 搜索内容
返回:
- 包含搜索结果的字典
"""
logger.info(f"执行网络搜索: {query}")
if not WEB_WEBPILOT_APIKEY:
return {
"success": False,
"error": "未设置WebPilot APIKEY"
}
try:
# 构造一个可能包含搜索关键词的URL (使用必应搜索)
url = f"https://www.bing.com/search?q={query}"
# 准备请求参数
payload = {
"link": url,
"ur": query,
"lp": True,
"rt": False,
"l": "zh-CN",
}
# 设置请求头
headers = {
'Content-Type': 'application/json',
'WebPilot-Friend-UID': WEB_WEBPILOT_APIKEY
}
# 发送请求
response = requests.post(API_URL, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
return {
"success": True,
"result": data
}
except Exception as e:
logger.error(f"搜索失败: {str(e)}")
return {
"success": False,
"result": str(e)
}
@mcp.tool()
def read_webpage(url: str, keyword: str = "", language: str = "zh-CN") -> dict:
"""
读取并分析网页内容。
参数:
- url: 要读取的网页URL
- keyword: 在网页中查找的关键词(可选)
- language: 语言代码，默认为中文
返回:
- 包含网页内容的字典
"""
logger.info(f"读取网页: {url}, 关键词: {keyword}")
if not WEB_WEBPILOT_APIKEY:
return {
"success": False,
"error": "未设置WebPilot APIKEY"
}
try:
# 准备请求参数
payload = {
"link": url,
"ur": keyword,
"lp": True,
"rt": False,
"l": language
}
# 设置请求头
headers = {
'Content-Type': 'application/json',
'WebPilot-Friend-UID': WEB_WEBPILOT_APIKEY
}
# 发送请求
response = requests.post(API_URL, json=payload, headers=headers)
response.raise_for_status()
data = response.json()
logger.info(f"成功读取网页，标题: {data.get('title', '无标题')}")
return {
"success": True,
"title": data.get("title", ""),
"content": data.get("content", ""),
"url": url
}
except Exception as e:
logger.error(f"读取网页失败: {str(e)}")
return {
"success": False,
"error": str(e)
}