tanjianbin/lzwcai-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 新增飞书和文件工具MCP服务并优化机器人配置

Browse Source 
- 新增飞书MCP服务，包含图片上传和消息卡片发送功能
- 新增文件工具MCP服务，提供文件转JSON和data URI功能
- 更新机器人MCP服务配置，移除环境变量硬编码
- 升级机器人版本至0.1.28并优化迎宾功能参数
- 添加.gitignore和项目配置文件
This commit is contained in:
Sucan
2026-02-11 18:57:31 +08:00
parent ff39bdbd8a
commit 0a308726a6
20 changed files with 1005 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
__pycache__/

117
file_tools/README.md Normal file
View File

@@ -0,0 +1,117 @@
# File Tools MCP 技能包
该技能包提供文件与 URL 的 Base64 处理能力，主要用于将文件内容转换为 JSON 字符串或 data URI，供 MCP Agent 直接调用。
## 功能
- 文件对象转 JSON 字符串
- 文件对象转 data URI（Base64）
- 文件路径转 data URI（Base64）
- 文件 URL 转 data URI（Base64）
## 工具列表
### 1. file_to_json
将文件对象转储为 JSON 字符串。
**入参**
```json
{
"file": {
"name": "example.txt",
"type": "text/plain",
"size": 123,
"last_modified": 1700000000,
"content_base64": "SGVsbG8="
}
}
```
也可以使用 `file_path`：
```json
{
"file_path": "e:\\lzwcai-mcp\\README.md"
}
```
### 2. file_to_data_uri
将文件对象转换为 data URI。
**入参**
```json
{
"file": {
"name": "hello.txt",
"content_base64": "SGVsbG8="
},
"type": "text/plain"
}
```
### 3. file_path_to_data_uri
将本地文件路径转换为 data URI。
**入参**
```json
{
"file_path": "e:\\lzwcai-mcp\\README.md",
"type": "text/plain"
}
```
### 4. url_to_data_uri
将文件 URL 转换为 data URI。
**入参**
```json
{
"url": "https://example.com/image.png"
}
```
## 启动方式
### uvx
```json
{
"mcpServers": {
"file-tools-mcp": {
"command": "uvx",
"type": "stdio",
"args": [
"lzwcai-mcpskills-file-tools-mcp"
],
"env": {}
}
}
}
```
### python
```json
{
"mcpServers": {
"file-tools-mcp": {
"disabled": false,
"type": "stdio",
"timeout": 30,
"command": "python",
"args": [
"-m",
"file_tools.main"
]
}
}
}
```

1
file_tools/file_tools/__init__.py Normal file
View File

@@ -0,0 +1 @@

47
file_tools/file_tools/main.py Normal file
View File

@@ -0,0 +1,47 @@
import asyncio
import logging
from mcp.server import Server
from mcp.server.stdio import stdio_server
from .tools import handle_call_tool, tools
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
class FileToolsServer:
def __init__(self) -> None:
self.server: Server = Server("lzwcai-mcpskills-file-tools-mcp")
self._register_handlers()
def _register_handlers(self) -> None:
@self.server.list_tools()
async def list_tools():
return tools
@self.server.call_tool()
async def call_tool(name: str, arguments: dict):
logger.info("Received tool call request: %s", name)
return await handle_call_tool(name, arguments)
async def _main() -> None:
server = FileToolsServer()
async with stdio_server() as (read_stream, write_stream):
await server.server.run(
read_stream,
write_stream,
server.server.create_initialization_options()
)
def main() -> None:
asyncio.run(_main())
if __name__ == "__main__":
main()

256
file_tools/file_tools/tools.py Normal file
View File

@@ -0,0 +1,256 @@
import base64
import json
import mimetypes
import tempfile
from pathlib import Path
from urllib.parse import urlparse
from typing import Any, Dict, List, Optional, Tuple
import httpx
from mcp.types import TextContent, Tool
tools: List[Tool] = [
Tool(
name="file_to_json",
description="将文件对象转储为 JSON 字符串",
inputSchema={
"type": "object",
"properties": {
"file": {
"type": "object",
"properties": {
"name": {"type": "string"},
"type": {"type": "string"},
"size": {"type": "integer"},
"last_modified": {"type": "integer"},
"content_base64": {"type": "string"}
},
"required": ["name"]
},
"file_path": {"type": "string"}
},
"required": []
}
),
Tool(
name="file_to_data_uri",
description="将文件转换为 data URI(Base64 编码)",
inputSchema={
"type": "object",
"properties": {
"file": {
"type": "object",
"properties": {
"name": {"type": "string"},
"type": {"type": "string"},
"size": {"type": "integer"},
"last_modified": {"type": "integer"},
"content_base64": {"type": "string"}
},
"required": ["name", "content_base64"]
},
"type": {"type": "string"},
"file_path": {"type": "string"}
},
"required": []
}
),
Tool(
name="file_path_to_data_uri",
description="将文件路径转换为 data URI(Base64 编码)",
inputSchema={
"type": "object",
"properties": {
"file_path": {"type": "string"},
"type": {"type": "string"}
},
"required": ["file_path"]
}
),
Tool(
name="url_to_data_uri",
description="将文件 URL 转换为 data URI(Base64 编码)",
inputSchema={
"type": "object",
"properties": {
"url": {"type": "string"}
},
"required": ["url"]
}
),
Tool(
name="url_to_temp_file",
description="将文件 URL 下载为临时文件并返回路径",
inputSchema={
"type": "object",
"properties": {
"url": {"type": "string"},
"file_path": {"type": "string"}
},
"required": ["url"]
}
)
]
def _guess_mime_type(name: Optional[str]) -> str:
if not name:
return "application/octet-stream"
mime_type, _ = mimetypes.guess_type(name)
return mime_type or "application/octet-stream"
def _is_url(value: Optional[str]) -> bool:
if not value:
return False
parsed = urlparse(value)
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
def _decode_base64(content_base64: str) -> bytes:
try:
return base64.b64decode(content_base64, validate=True)
except Exception:
return base64.b64decode(content_base64)
def _build_data_uri(data: bytes, mime_type: str) -> str:
encoded = base64.b64encode(data).decode("ascii")
return f"data:{mime_type};base64,{encoded}"
def _extract_file_payload(arguments: Dict[str, Any]) -> Tuple[Dict[str, Any], Optional[bytes], Optional[str]]:
file_payload = arguments.get("file")
file_path = arguments.get("file_path")
if file_payload is None and file_path is None:
raise ValueError("missing file or file_path")
name = None
mime_type = None
size = None
last_modified = None
content_base64 = None
data = None
if file_payload is not None:
name = file_payload.get("name")
mime_type = file_payload.get("type")
size = file_payload.get("size")
last_modified = file_payload.get("last_modified")
content_base64 = file_payload.get("content_base64")
if file_path:
path = Path(file_path)
if name is None:
name = path.name
if content_base64 is None:
data = path.read_bytes()
if data is None and content_base64 is not None:
data = _decode_base64(content_base64)
return (
{
"name": name,
"type": mime_type,
"size": size,
"last_modified": last_modified,
"content_base64": content_base64
},
data,
name
)
def _normalize_mime_type(mime_type: Optional[str], name: Optional[str]) -> str:
if mime_type:
return mime_type
return _guess_mime_type(name)
def _build_file_json(arguments: Dict[str, Any]) -> str:
file_info, data, name = _extract_file_payload(arguments)
if file_info["type"] is None:
file_info["type"] = _guess_mime_type(name)
if file_info["size"] is None and data is not None:
file_info["size"] = len(data)
return json.dumps(file_info, ensure_ascii=False)
def _build_file_data_uri(arguments: Dict[str, Any]) -> str:
file_info, data, name = _extract_file_payload(arguments)
if data is None:
raise ValueError("missing file content for data uri")
mime_type = arguments.get("type") or file_info.get("type")
mime_type = _normalize_mime_type(mime_type, name)
return _build_data_uri(data, mime_type)
def _build_path_data_uri(arguments: Dict[str, Any]) -> str:
file_path = arguments.get("file_path")
if not file_path:
raise ValueError("missing file_path")
path = Path(file_path)
data = path.read_bytes()
mime_type = arguments.get("type") or _guess_mime_type(path.name)
return _build_data_uri(data, mime_type)
async def _build_url_data_uri(arguments: Dict[str, Any]) -> str:
url = arguments.get("url")
if not url:
raise ValueError("missing url")
async with httpx.AsyncClient(follow_redirects=True, timeout=20) as client:
response = await client.get(url)
response.raise_for_status()
content_type = response.headers.get("content-type", "")
mime_type = content_type.split(";")[0].strip() if content_type else ""
if not mime_type:
mime_type = _guess_mime_type(url)
return _build_data_uri(response.content, mime_type)
async def _build_url_file_path(arguments: Dict[str, Any]) -> str:
url = arguments.get("url")
if not url:
raise ValueError("missing url")
file_path = arguments.get("file_path")
if file_path:
path = Path(file_path)
path.parent.mkdir(parents=True, exist_ok=True)
else:
suffix = Path(urlparse(url).path).suffix
with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as temp_file:
path = Path(temp_file.name)
async with httpx.AsyncClient(follow_redirects=True, timeout=20) as client:
response = await client.get(url)
response.raise_for_status()
path.write_bytes(response.content)
return str(path)
async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
try:
if name == "file_to_json":
result = _build_file_json(arguments)
elif name == "file_to_data_uri":
file_path = arguments.get("file_path")
if _is_url(file_path):
result = await _build_url_data_uri({"url": file_path})
else:
result = _build_file_data_uri(arguments)
elif name == "file_path_to_data_uri":
file_path = arguments.get("file_path")
if _is_url(file_path):
result = await _build_url_data_uri({"url": file_path})
else:
result = _build_path_data_uri(arguments)
elif name == "url_to_data_uri":
result = await _build_url_data_uri(arguments)
elif name == "url_to_temp_file":
result = await _build_url_file_path(arguments)
else:
raise ValueError(f"unknown tool name: {name}")
return [TextContent(type="text", text=result)]
except Exception as exc:
return [TextContent(type="text", text=f"Failed to call tool {name}: {exc}")]

14
file_tools/mcp-server-file-tools.json Normal file
View File

@@ -0,0 +1,14 @@
{
"mcpServers": {
"lzwcai-mcpskills-file-tools-mcp": {
"disabled": false,
"type": "stdio",
"timeout": 30,
"command": "uvx",
"args": [
"lzwcai-mcpskills-file-tools-mcp"
],
"env": {}
}
}
}

20
file_tools/pyproject.toml Normal file
View File

@@ -0,0 +1,20 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "lzwcai-mcpskills-file-tools-mcp"
version = "0.1.9"
description = "File tools MCP server"
requires-python = ">=3.10"
dependencies = [
"mcp[cli]>=1.6.0",
"httpx>=0.24.0"
]
[project.scripts]
lzwcai-mcpskills-file-tools-mcp = "file_tools.main:main"
[tool.hatch.build.targets.wheel]
packages = ["file_tools"]

67
file_tools/test_tools.py Normal file
View File

@@ -0,0 +1,67 @@
import argparse
import asyncio
import base64
from typing import Optional
from file_tools.tools import handle_call_tool
async def _run(
url: Optional[str],
file_path: Optional[str],
text: Optional[str],
name: str,
download_url: Optional[str],
download_path: Optional[str]
) -> None:
if url:
result = await handle_call_tool("url_to_data_uri", {"url": url})
print("url_to_data_uri:", result[0].text[:80], "len=", len(result[0].text))
if file_path:
result = await handle_call_tool("file_path_to_data_uri", {"file_path": file_path})
print("file_path_to_data_uri:", result[0].text[:80], "len=", len(result[0].text))
if download_url:
payload = {"url": download_url}
if download_path:
payload["file_path"] = download_path
result = await handle_call_tool("url_to_temp_file", payload)
print("url_to_temp_file:", result[0].text)
if text is not None:
payload = {
"file": {
"name": name,
"content_base64": base64.b64encode(text.encode("utf-8")).decode("ascii")
}
}
result = await handle_call_tool("file_to_json", payload)
print("file_to_json:", result[0].text)
result = await handle_call_tool("file_to_data_uri", payload)
print("file_to_data_uri:", result[0].text[:80], "len=", len(result[0].text))
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--url")
parser.add_argument("--file-path")
parser.add_argument("--download-url")
parser.add_argument("--download-path")
parser.add_argument("--text")
parser.add_argument("--name", default="sample.txt")
args = parser.parse_args()
asyncio.run(
_run(
args.url,
args.file_path,
args.text,
args.name,
args.download_url,
args.download_path
)
)
if __name__ == "__main__":
main()

1
lzwcai_lark_mcp/lzwcai_lark_mcp/__init__.py Normal file
View File

@@ -0,0 +1 @@
__all__ = ["main"]

9
lzwcai_lark_mcp/lzwcai_lark_mcp/config.py Normal file
View File

@@ -0,0 +1,9 @@
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
APP_ID = os.getenv("app_id", "")
APP_SECRET = os.getenv("app_secret", "")

85
lzwcai_lark_mcp/lzwcai_lark_mcp/main.py Normal file
View File

@@ -0,0 +1,85 @@
import asyncio
import logging
import os
import time
from typing import Tuple
import requests
from mcp.server import Server
from mcp.server.stdio import stdio_server
from .config import Config
from .tools import handle_call_tool, tools
logging.basicConfig(
level=getattr(logging, os.getenv("LOG_LEVEL", "INFO"), logging.INFO),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
class LarkMcpServer:
def __init__(self) -> None:
self.server = Server("lzwcai-mcpskills-lark-mcp")
self.tenant_access_token: str | None = None
self.token_expire_at: float = 0
self._register_handlers()
def _register_handlers(self) -> None:
@self.server.list_tools()
async def list_tools():
return tools
@self.server.call_tool()
async def call_tool(name: str, arguments: dict):
await self.ensure_token()
return await handle_call_tool(name, arguments, self.tenant_access_token or "")
async def ensure_token(self) -> None:
if self.tenant_access_token and time.time() < self.token_expire_at - 60:
return
token, expires_in = await asyncio.to_thread(self._request_token)
self.tenant_access_token = token
self.token_expire_at = time.time() + expires_in
def _request_token(self) -> Tuple[str, int]:
if not Config.APP_ID or not Config.APP_SECRET:
raise ValueError("app_id or app_secret is missing")
auth_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
headers = {"Content-Type": "application/json"}
payload = {"app_id": Config.APP_ID, "app_secret": Config.APP_SECRET}
response = requests.post(
auth_url,
json=payload,
headers=headers,
timeout=10
)
response.raise_for_status()
data = response.json()
if data.get("code") not in (0, None):
raise RuntimeError(f"lark auth failed: {data}")
token = data.get("tenant_access_token")
if not token:
raise RuntimeError(f"lark auth response missing token: {data}")
expires_in = int(data.get("expire", 3600))
return token, expires_in
async def _main() -> None:
server = LarkMcpServer()
await server.ensure_token()
async with stdio_server() as (read_stream, write_stream):
await server.server.run(
read_stream,
write_stream,
server.server.create_initialization_options()
)
def main() -> None:
asyncio.run(_main())
if __name__ == "__main__":
main()

280
lzwcai_lark_mcp/lzwcai_lark_mcp/tools.py Normal file
View File

@@ -0,0 +1,280 @@
import json
import mimetypes
import os
import re
from datetime import datetime
from typing import Dict, List
from urllib.parse import urlparse
import requests
from mcp.types import Tool, TextContent
tools = [
Tool(
name="upload_image_by_url",
description="上传图片并返回image_key，入参为可下载的文件URL",
inputSchema={
"type": "object",
"properties": {
"image_type": {
"type": "string",
"description": "图片类型，例如 message"
},
"url": {
"type": "string",
"description": "可直接下载的图片文件URL"
}
},
"required": ["url"]
}
),
Tool(
name="send_card_message",
description="发送消息卡片，入参为receiver_ids、person_id和image_key",
inputSchema={
"type": "object",
"properties": {
"receiver_ids": {
"type": "array",
"description": "消息接收者ID列表",
"items": {
"type": "string"
}
},
"person_id": {
"type": "string",
"description": "卡片内person组件的ID"
},
"image_key": {
"type": "string",
"description": "图片image_key"
}
},
"required": ["image_key", "person_id"]
}
)
]
def upload_image_by_url(token: str, url: str, image_type: str) -> str:
download_response = requests.get(url, stream=True, timeout=30)
download_response.raise_for_status()
content = download_response.content
content_disposition = download_response.headers.get("Content-Disposition") or download_response.headers.get("content-disposition")
filename = ""
if content_disposition:
match = re.search(r'filename\*?=(?:UTF-8\'\')?"?([^";]+)"?', content_disposition)
if match:
filename = match.group(1).split("/")[-1]
if not filename:
path = urlparse(url).path
filename = path.split("/")[-1] if path else ""
if not filename:
filename = "image"
content_type = download_response.headers.get("Content-Type")
if not content_type:
content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
headers = {"Authorization": f"Bearer {token}"}
data = {"image_type": image_type}
files = {"image": (filename, content, content_type)}
response = requests.post(
"https://open.feishu.cn/open-apis/im/v1/images",
headers=headers,
data=data,
files=files,
timeout=30
)
response.raise_for_status()
payload = response.json()
if payload.get("code") not in (0, None):
raise RuntimeError(f"lark image upload failed: {payload}")
image_key = payload.get("data", {}).get("image_key")
if not image_key:
raise RuntimeError(f"lark image upload missing image_key: {payload}")
return image_key
def send_card_message(token: str, receiver_id: str, person_id: str, image_key: str) -> str:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
content = {
"schema": "2.0",
"config": {
"update_multi": True,
"locales": ["en_us"],
"style": {
"text_size": {
"normal_v2": {
"default": "normal",
"pc": "normal",
"mobile": "heading"
}
}
}
},
"body": {
"direction": "vertical",
"padding": "12px 12px 12px 12px",
"elements": [
{
"tag": "markdown",
"content": f"<font color=\"grey\">时间</font> {current_time}",
"i18n_content": {
"en_us": f"<font color=\"grey\">Incident time</font>\n{current_time}"
},
"text_align": "center",
"text_size": "normal_v2",
"margin": "0px 0px 0px 0px"
},
{
"tag": "markdown",
"content": f"\n<person id='{person_id}' show_name=true show_avatar=true style='normal'></person><font color=\"grey\">进入仓库</font>",
"i18n_content": {
"en_us": "<font color=\"grey\">Alert details</font>\nMobile client crash rate at 5%"
},
"text_align": "center",
"text_size": "normal",
"margin": "0px 0px 0px 0px"
},
{
"tag": "column_set",
"horizontal_spacing": "8px",
"horizontal_align": "left",
"columns": [
{
"tag": "column",
"width": "weighted",
"elements": [],
"vertical_spacing": "8px",
"horizontal_align": "left",
"vertical_align": "top",
"weight": 1
},
{
"tag": "column",
"width": "auto",
"elements": [
{
"tag": "img",
"img_key": image_key,
"preview": True,
"transparent": False,
"scale_type": "crop_center",
"size": "large",
"corner_radius": "16px",
"margin": "0px 0px 0px 0px"
}
],
"vertical_spacing": "8px",
"horizontal_align": "left",
"vertical_align": "top"
},
{
"tag": "column",
"width": "weighted",
"elements": [],
"vertical_spacing": "8px",
"horizontal_align": "left",
"vertical_align": "top",
"weight": 1
}
],
"margin": "0px 0px 0px 0px"
},
{
"tag": "hr",
"margin": "0px 0px 0px 0px"
}
]
},
"header": {
"title": {
"tag": "plain_text",
"content": "人员进入仓库通知",
"i18n_content": {
"en_us": "[Action Needed] Alert: Process Error - Please Address Promptly"
}
},
"subtitle": {
"tag": "plain_text",
"content": ""
},
"template": "wathet",
"icon": {
"tag": "standard_icon",
"token": "bell_filled"
},
"padding": "12px 12px 12px 12px"
}
}
payload = {
"receive_id": receiver_id,
"msg_type": "interactive",
"content": json.dumps(content, ensure_ascii=False)
}
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
response = requests.post(
"https://open.feishu.cn/open-apis/im/v1/messages",
params={"receive_id_type": "user_id"},
headers=headers,
json=payload,
timeout=30
)
try:
data = response.json()
except ValueError:
data = {"raw": response.text}
if not response.ok:
raise RuntimeError(f"lark send message http error: {data}")
if data.get("code") not in (0, None):
raise RuntimeError(f"lark send message failed: {data}")
message_id = data.get("data", {}).get("message_id")
if not message_id:
raise RuntimeError(f"lark send message missing message_id: {data}")
return message_id
def send_card_messages(token: str, receiver_ids: List[str], person_id: str, image_key: str) -> str:
if not receiver_ids:
raise ValueError("missing receiver_ids")
message_ids = []
for receiver_id in receiver_ids:
normalized_receiver_id = str(receiver_id).strip()
if not normalized_receiver_id:
continue
message_ids.append(send_card_message(token, normalized_receiver_id, person_id, image_key))
if not message_ids:
raise ValueError("missing receiver_ids")
return json.dumps(message_ids, ensure_ascii=False)
async def handle_call_tool(name: str, arguments: Dict[str, object], token: str) -> List[TextContent]:
try:
if name == "upload_image_by_url":
image_type = str(arguments.get("image_type") or "message").strip()
url = str(arguments.get("url", "")).strip()
if not image_type:
raise ValueError("missing image_type")
if not url:
raise ValueError("missing url")
result = upload_image_by_url(token, url, image_type)
elif name == "send_card_message":
image_key = str(arguments.get("image_key", "")).strip()
receiver_ids = arguments.get("receiver_ids")
person_id = str(arguments.get("person_id", "")).strip()
if not image_key:
raise ValueError("missing image_key")
if not person_id:
raise ValueError("missing person_id")
if receiver_ids is not None:
if not isinstance(receiver_ids, list):
raise ValueError("receiver_ids must be a list")
result = send_card_messages(token, receiver_ids, person_id, image_key)
else:
raise ValueError(f"unknown tool name: {name}")
return [TextContent(type="text", text=result)]
except Exception as exc:
return [TextContent(type="text", text=f"Failed to call tool {name}: {exc}")]

17
lzwcai_lark_mcp/mcp-server.json Normal file
View File

@@ -0,0 +1,17 @@
{
"mcpServers": {
"lzwcai-mcpskills-lark-mcp": {
"disabled": false,
"type": "stdio",
"timeout": 30,
"command": "uvx",
"args": [
"lzwcai-mcpskills-lark-mcp"
],
"env": {
"app_id": "cli_a8d0e0c140169013",
"app_secret": "yEc0E8Aoo8Mo9NPPzphidez51xB71HXW"
}
}
}
}

20
lzwcai_lark_mcp/pyproject.toml Normal file
View File

@@ -0,0 +1,20 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "lzwcai-mcpskills-lark-mcp"
version = "0.1.4"
description = "Lark MCP server"
requires-python = ">=3.10"
dependencies = [
"mcp[cli]>=1.6.0",
"python-dotenv>=0.21.0",
"requests>=2.25.0"
]
[project.scripts]
lzwcai-mcpskills-lark-mcp = "lzwcai_lark_mcp.main:main"
[tool.hatch.build.targets.wheel]
packages = ["lzwcai_lark_mcp"]

38
lzwcai_lark_mcp/test.py Normal file
View File

@@ -0,0 +1,38 @@
import asyncio
import json
import os
from pathlib import Path
def main() -> None:
if not os.getenv("app_id") or not os.getenv("app_secret"):
config_path = Path(__file__).with_name("mcp-server.json")
if config_path.exists():
config_data = json.loads(config_path.read_text(encoding="utf-8"))
env_data = (
config_data.get("mcpServers", {})
.get("lzwcai-mcpskills-lark-mcp", {})
.get("env", {})
)
if env_data.get("app_id") and env_data.get("app_secret"):
os.environ["app_id"] = env_data["app_id"]
os.environ["app_secret"] = env_data["app_secret"]
if not os.getenv("app_id") or not os.getenv("app_secret"):
raise RuntimeError("missing app_id or app_secret")
from lzwcai_lark_mcp.main import LarkMcpServer
from lzwcai_lark_mcp.tools import send_card_message
async def _run() -> None:
server = LarkMcpServer()
await server.ensure_token()
image_key = "img_v3_02uq_d48b3ee1-0f89-44a3-80cd-a5afa4f8c39g"
receiver_id = "843ga2gb"
person_id = receiver_id
result = send_card_message(server.tenant_access_token or "", receiver_id, person_id, image_key)
print(result)
asyncio.run(_run())
if __name__ == "__main__":
main()

BIN
terminal_temi_mcp/dist/terminal_temi_mcp-0.1.21-py3-none-any.whl vendored
View File

Binary file not shown.

BIN
terminal_temi_mcp/dist/terminal_temi_mcp-0.1.21.tar.gz vendored
View File

Binary file not shown.

8
terminal_temi_mcp/mcp-server-temi.json
View File

@@ -1,17 +1,13 @@
{
"mcpServers": {
"terminal-temi-mcp": {
"terminal_temi_mcp": {
"disabled": false,
"type": "stdio",
"timeout": 30,
"command": "uvx",
"args": [
"terminal_temi_mcp"
],
"env": {
"employeeId": "$employeeId$",
"userId": "$employeeId$"
}
]
}
}
}

2
terminal_temi_mcp/pyproject.toml
View File

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "terminal_temi_mcp"
version = "0.1.21"
version = "0.1.28"
description = "MQTT-based navigation server for robot"
requires-python = ">=3.10"
dependencies = [

44
terminal_temi_mcp/terminal_temi_mcp/main.py
View File

@@ -50,7 +50,7 @@ class NavServer:
params = {
"data": {"location": location, "flag": flag}
}
return await self.publish_Cmd("123456", user_id, "nav", params)
return await self.publish_Cmd("111111", user_id, "nav", params)
except Exception as e:
logger.error(f"Failed to call navigation mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call navigation mcp-tool: {str(e)}"
@@ -64,12 +64,12 @@ class NavServer:
params = {
"data": {"speech": speech}
}
return await self.publish_Cmd("123456", user_id, "speak", params)
return await self.publish_Cmd("111111", user_id, "speak", params)
except Exception as e:
logger.error(f"Failed to call speak mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call speak mcp-tool: {str(e)}"
async def reception(self, user_id: str, location: str):
async def reception(self, user_id: str, location: str, name: str = "贵宾", waitingTime: int = 20*60*1000):
"""轮足机器人移动到指定位置迎宾"""
try:
if not location:
@@ -77,10 +77,12 @@ class NavServer:
else:
params = {
"data": {
"location": location
"location": location,
"name": name,
"waitingTime": waitingTime
}
}
return await self.publish_Cmd("123456", user_id, "reception", params)
return await self.publish_Cmd("111111", user_id, "reception", params)
except Exception as e:
logger.error(f"Failed to call reception mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call reception mcp-tool: {str(e)}"
@@ -97,7 +99,7 @@ class NavServer:
"text": text
}
}
return await self.publish_Cmd("123456", user_id, "notification", params)
return await self.publish_Cmd("111111", user_id, "notification", params)
except Exception as e:
logger.error(f"Failed to call notification mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call notification mcp-tool: {str(e)}"
@@ -110,7 +112,7 @@ class NavServer:
"action": "repose"
}
}
return await self.publish_Cmd("123456", user_id, "map", params)
return await self.publish_Cmd("111111", user_id, "map", params)
except Exception as e:
logger.error(f"Failed to call repose mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call repose mcp-tool: {str(e)}"
@@ -127,7 +129,7 @@ class NavServer:
"next_location": next_location
}
}
return await self.publish_Cmd("123456", user_id, "delivery", params)
return await self.publish_Cmd("111111", user_id, "delivery", params)
except Exception as e:
logger.error(f"Failed to call delivery mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call delivery mcp-tool: {str(e)}"
@@ -141,7 +143,7 @@ class NavServer:
params = {
"data": {"locations": locations}
}
return await self.publish_Cmd("123456", user_id, "patrol", params)
return await self.publish_Cmd("111111", user_id, "patrol", params)
except Exception as e:
logger.error(f"Failed to call patrol mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call patrol mcp-tool: {str(e)}"
@@ -155,7 +157,7 @@ class NavServer:
params = {
"data": {}
}
return await self.publish_Cmd("123456", user_id, "face", params)
return await self.publish_Cmd("111111", user_id, "face", params)
except Exception as e:
logger.error(f"Failed to call face recognize mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call face recognize mcp-tool: {str(e)}"
@@ -169,7 +171,7 @@ class NavServer:
params = {
"data": {}
}
return await self.publish_Cmd("123456", user_id, "qrcode", params)
return await self.publish_Cmd("111111", user_id, "qrcode", params)
except Exception as e:
logger.error(f"Failed to call QR code scan mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call QR code scan mcp-tool: {str(e)}"
@@ -183,7 +185,7 @@ class NavServer:
params = {
"data": {"location_name": location_name}
}
return await self.publish_Cmd("123456", user_id, "saveLocation", params)
return await self.publish_Cmd("111111", user_id, "saveLocation", params)
except Exception as e:
logger.error(f"Failed to call save_position mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call save_position mcp-tool: {str(e)}"
@@ -196,7 +198,7 @@ class NavServer:
params = {
"data": [data.model_dump() for data in datas]
}
return await self.publish_Cmd("123456", user_id, "guide", params)
return await self.publish_Cmd("111111", user_id, "guide", params)
except Exception as e:
logger.error(f"Failed to call guide mcp-tool: {str(e)} ", exc_info=True)
return f"Failed to call guide mcp-tool: {str(e)}"
@@ -263,12 +265,22 @@ async def serve() -> None:
"description": "引导接待客人到这个位置",
"minLength": 1
},
"name": {
"type": "string",
"description": "客人姓名",
"minLength": 1
},
"waitingTime": {
"type": "integer",
"description": "等待时间(单位:毫秒)",
"default": 20*60*1000
},
"user_id": {
"type": "string",
"description": "用户ID"
}
},
"required": ["location", "user_id"]
"required": ["location", "name", "user_id"]
}
),
# Tool(
@@ -437,7 +449,9 @@ async def serve() -> None:
raise ValueError("缺少必要参数: location or user_id")
result = await nav_server.reception(
user_id=arguments["user_id"],
location=arguments["location"]
location=arguments["location"],
name=arguments.get("name", ""),
waitingTime=arguments.get("waitingTime", 20*60*1000)
)
# elif name == "notification":
# if "location" not in arguments or "text" not in arguments or "user_id" not in arguments: