"""
飞书 MCP 服务器 - 使用官方 MCP SDK 实现
"""
import asyncio
import json
from typing import Any
import lark_oapi as lark
import mcp.server.stdio
import mcp.types as types
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
from yuppie_mcp_feishu.impl import (
# 应用级工具
create_bitable_app_impl,
copy_bitable_app_impl,
create_bitable_table_impl,
# 记录级工具
create_bitable_record_impl,
update_bitable_record_impl,
search_bitable_records_impl,
batch_create_bitable_records_impl,
batch_update_bitable_records_impl,
batch_get_bitable_records_impl,
batch_delete_bitable_records_impl,
)
server = Server("yuppie-mcp-feishu")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
列出所有可用的 MCP 工具
"""
return [
# ===== 应用级工具 (3个) =====
types.Tool(
name="create_bitable_app",
description="创建多维表格应用\n\n"
"参数:\n"
" name: 多维表格名称\n"
" folder_token: (可选) 文件夹token，指定创建位置\n\n"
"返回:\n"
" JSON格式的应用信息，包含app_token等",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string"},
"folder_token": {"type": "string"},
},
"required": ["name"],
},
),
types.Tool(
name="copy_bitable_app",
description="复制多维表格应用\n\n"
"参数:\n"
" app_token: 要复制的多维表格token\n"
" name: 新多维表格的名称\n"
" folder_token: (可选) 文件夹token\n\n"
"返回:\n"
" JSON格式的应用信息",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"name": {"type": "string"},
"folder_token": {"type": "string"},
},
"required": ["app_token", "name"],
},
),
types.Tool(
name="create_bitable_table",
description="创建数据表\n\n"
"参数:\n"
" app_token: 多维表格的app_token\n"
" table_name: 数据表名称\n"
" fields: (可选) 字段定义列表的JSON字符串，例如: [{\"type\":1,\"name\":\"字段名\"}]\n\n"
"返回:\n"
" JSON格式的数据表信息，包含table_id等",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_name": {"type": "string"},
"fields": {"type": "string"},
},
"required": ["app_token", "table_name"],
},
),
# ===== 记录级工具 (7个) =====
types.Tool(
name="create_bitable_record",
description="在多维表格中创建单条记录\n\n"
"参数:\n"
" app_token: 多维表格的token\n"
" table_id: 数据表ID\n"
" fields: 记录字段字典的JSON字符串，例如: {\"字段名\":\"值\"}\n\n"
"返回:\n"
" 创建的记录信息",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_id": {"type": "string"},
"fields": {"type": "string"},
},
"required": ["app_token", "table_id", "fields"],
},
),
types.Tool(
name="update_bitable_record",
description="更新多维表格中的单条记录\n\n"
"参数:\n"
" app_token: 多维表格的token\n"
" table_id: 数据表ID\n"
" record_id: 记录ID\n"
" fields: 要更新的字段字典的JSON字符串，例如: {\"字段名\":\"新值\"}\n\n"
"返回:\n"
" 更新后的记录信息",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_id": {"type": "string"},
"record_id": {"type": "string"},
"fields": {"type": "string"},
},
"required": ["app_token", "table_id", "record_id", "fields"],
},
),
types.Tool(
name="search_bitable_records",
description="查询多维表格记录（支持分页）\n\n"
"参数:\n"
" app_token: 多维表格的token\n"
" table_id: 数据表ID\n"
" page_size: 每页记录数（1-500，默认20）\n"
" page_token: 分页token，首次查询为空\n"
" filter: (可选) 过滤条件的JSON字符串，例如: {\"conjunction\":\"and\",\"conditions\":[...]}\n\n"
"返回:\n"
" 记录列表和分页信息",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_id": {"type": "string"},
"page_size": {"type": "integer", "default": 20},
"page_token": {"type": "string"},
"filter": {"type": "string"},
},
"required": ["app_token", "table_id"],
},
),
types.Tool(
name="batch_create_bitable_records",
description="批量创建多维表格记录（最多1000条）\n\n"
"参数:\n"
" app_token: 多维表格的token\n"
" table_id: 数据表ID\n"
" records: 记录列表的JSON字符串，例如: [{\"fields\":{\"字段名\":\"值\"}}]\n\n"
"返回:\n"
" 创建的记录信息",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_id": {"type": "string"},
"records": {"type": "string"},
},
"required": ["app_token", "table_id", "records"],
},
),
types.Tool(
name="batch_update_bitable_records",
description="批量更新多维表格记录（最多1000条）\n\n"
"参数:\n"
" app_token: 多维表格的token\n"
" table_id: 数据表ID\n"
" records: 要更新的记录列表的JSON字符串，例如: [{\"record_id\":\"xxx\",\"fields\":{\"字段名\":\"新值\"}}]\n\n"
"返回:\n"
" 更新后的记录信息",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_id": {"type": "string"},
"records": {"type": "string"},
},
"required": ["app_token", "table_id", "records"],
},
),
types.Tool(
name="batch_get_bitable_records",
description="批量获取多维表格记录（最多100条）\n\n"
"参数:\n"
" app_token: 多维表格的token\n"
" table_id: 数据表ID\n"
" record_ids: 记录ID列表的JSON字符串，例如: [\"id1\",\"id2\",\"id3\"]\n\n"
"返回:\n"
" 记录信息",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_id": {"type": "string"},
"record_ids": {"type": "string"},
},
"required": ["app_token", "table_id", "record_ids"],
},
),
types.Tool(
name="batch_delete_bitable_records",
description="批量删除多维表格记录\n\n"
"参数:\n"
" app_token: 多维表格的token\n"
" table_id: 数据表ID\n"
" record_ids: 要删除的记录ID列表的JSON字符串，例如: [\"id1\",\"id2\",\"id3\"]\n\n"
"返回:\n"
" 删除结果",
inputSchema={
"type": "object",
"properties": {
"app_token": {"type": "string"},
"table_id": {"type": "string"},
"record_ids": {"type": "string"},
},
"required": ["app_token", "table_id", "record_ids"],
},
),
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict[str, Any] | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
处理工具调用请求
这是所有工具执行的统一入口点，根据工具名称分发到对应的实现函数
"""
if not arguments:
arguments = {}
try:
response = None
# ===== 应用级工具路由 =====
if name == "create_bitable_app":
name_val = arguments.get("name")
folder_token = arguments.get("folder_token", "")
if not name_val:
raise ValueError("Missing required parameter: name")
response = create_bitable_app_impl(name_val, folder_token)
elif name == "copy_bitable_app":
app_token = arguments.get("app_token")
name_val = arguments.get("name")
folder_token = arguments.get("folder_token", "")
if not app_token or not name_val:
raise ValueError("Missing required parameters: app_token, name")
response = copy_bitable_app_impl(app_token, name_val, folder_token)
elif name == "create_bitable_table":
app_token = arguments.get("app_token")
table_name = arguments.get("table_name")
fields_str = arguments.get("fields")
if not app_token or not table_name:
raise ValueError("Missing required parameters: app_token, table_name")
# 解析 JSON 字符串为数组
fields = json.loads(fields_str) if fields_str else None
response = create_bitable_table_impl(app_token, table_name, fields)
# ===== 记录级工具路由 =====
elif name == "create_bitable_record":
app_token = arguments.get("app_token")
table_id = arguments.get("table_id")
fields_str = arguments.get("fields")
if not app_token or not table_id or not fields_str:
raise ValueError(
"Missing required parameters: app_token, table_id, fields"
)
# 解析 JSON 字符串为字典
fields = json.loads(fields_str)
response = create_bitable_record_impl(app_token, table_id, fields)
elif name == "update_bitable_record":
app_token = arguments.get("app_token")
table_id = arguments.get("table_id")
record_id = arguments.get("record_id")
fields_str = arguments.get("fields")
if not all([app_token, table_id, record_id, fields_str]):
raise ValueError(
"Missing required parameters: app_token, table_id, record_id, fields"
)
# 解析 JSON 字符串为字典
fields = json.loads(fields_str)
response = update_bitable_record_impl(app_token, table_id, record_id, fields)
elif name == "search_bitable_records":
app_token = arguments.get("app_token")
table_id = arguments.get("table_id")
page_size = arguments.get("page_size", 20)
page_token = arguments.get("page_token", "")
filter_str = arguments.get("filter")
if not app_token or not table_id:
raise ValueError("Missing required parameters: app_token, table_id")
# 解析 JSON 字符串为字典（可选参数）
filter_cond = json.loads(filter_str) if filter_str else None
response = search_bitable_records_impl(
app_token, table_id, page_size, page_token, filter_cond
)
elif name == "batch_create_bitable_records":
app_token = arguments.get("app_token")
table_id = arguments.get("table_id")
records_str = arguments.get("records")
if not app_token or not table_id or not records_str:
raise ValueError("Missing required parameters: app_token, table_id, records")
# 解析 JSON 字符串为数组
records = json.loads(records_str)
response = batch_create_bitable_records_impl(app_token, table_id, records)
elif name == "batch_update_bitable_records":
app_token = arguments.get("app_token")
table_id = arguments.get("table_id")
records_str = arguments.get("records")
if not app_token or not table_id or not records_str:
raise ValueError("Missing required parameters: app_token, table_id, records")
# 解析 JSON 字符串为数组
records = json.loads(records_str)
response = batch_update_bitable_records_impl(app_token, table_id, records)
elif name == "batch_get_bitable_records":
app_token = arguments.get("app_token")
table_id = arguments.get("table_id")
record_ids_str = arguments.get("record_ids")
if not app_token or not table_id or not record_ids_str:
raise ValueError(
"Missing required parameters: app_token, table_id, record_ids"
)
# 解析 JSON 字符串为数组
record_ids = json.loads(record_ids_str)
response = batch_get_bitable_records_impl(app_token, table_id, record_ids)
elif name == "batch_delete_bitable_records":
app_token = arguments.get("app_token")
table_id = arguments.get("table_id")
record_ids_str = arguments.get("record_ids")
if not app_token or not table_id or not record_ids_str:
raise ValueError(
"Missing required parameters: app_token, table_id, record_ids"
)
# 解析 JSON 字符串为数组
record_ids = json.loads(record_ids_str)
response = batch_delete_bitable_records_impl(app_token, table_id, record_ids)
else:
raise ValueError(f"Unknown tool: {name}")
# 统一错误处理
if not response.success():
error_data = {
"success": False,
"code": response.code,
"msg": response.msg,
"error": getattr(response, "error", None),
"log_id": response.get_log_id(),
}
return [
types.TextContent(
type="text", text=lark.JSON.marshal(error_data, indent=4)
)
]
# 成功返回数据
return [
types.TextContent(
type="text", text=lark.JSON.marshal(response.data, indent=4)
)
]
except json.JSONDecodeError as e:
# JSON 解析错误
error_data = {
"success": False,
"error": f"Invalid JSON format: {str(e)}",
"error_type": "JSONDecodeError",
}
return [
types.TextContent(type="text", text=lark.JSON.marshal(error_data, indent=4))
]
except ValueError as e:
# 参数验证错误
error_data = {"success": False, "error": str(e), "error_type": "ValidationError"}
return [
types.TextContent(type="text", text=lark.JSON.marshal(error_data, indent=4))
]
except Exception as e:
# 其他未预期的错误
error_data = {
"success": False,
"error": str(e),
"error_type": "UnexpectedError",
}
return [
types.TextContent(type="text", text=lark.JSON.marshal(error_data, indent=4))
]
async def main():
"""
启动 MCP 服务器
使用 stdio 传输与客户端（如 Claude Desktop）通信
"""
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="yuppie-mcp-feishu",
server_version="0.5.1",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
