def create_server(client: FeishuClient | None = None) -> Any:
r"""
创建 OpenFeishu MCP 服务器。
Args:
client: 可选基础 [feishu.client.FeishuClient][]。未传入时会从环境变量构造：
`FEISHU_APP_ID` / `APP_ID`、`FEISHU_APP_SECRET` / `APP_SECRET`，
以及 `FEISHU_REGION` / `REGION`。
Returns:
已注册飞书工具的 `FastMCP` 实例。
Raises:
RuntimeError: 未安装 `open-feishu[mcp]` 额外依赖时抛出。
"""
try:
from mcp.server.fastmcp import FastMCP
except ImportError as exc: # pragma: no cover - depends on optional extra
raise RuntimeError("Install open-feishu[mcp] to use the MCP server") from exc
mcp = FastMCP("open-feishu", instructions=INSTRUCTIONS.strip())
base_client = client
def get_client(user_access_token: str | None = None) -> FeishuClient:
nonlocal base_client
if base_client is None:
base_client = FeishuClient(
_env("FEISHU_APP_ID", "APP_ID"),
_env("FEISHU_APP_SECRET", "APP_SECRET"),
region=_env("FEISHU_REGION", "REGION", default="feishu") or "feishu",
)
if user_access_token:
return base_client.as_user(user_access_token)
return base_client
@mcp.tool()
def feishu_parse_document_reference(text: str) -> dict[str, str | None]:
"""从 URL 或文本中提取飞书文档 / 知识库 token 与类型。"""
reference = parse_document_reference(text)
if reference is None:
return {"token": None, "doc_type": None}
return {"token": reference.token, "doc_type": reference.doc_type}
@mcp.tool()
async def feishu_authorize_url(
redirect_uri: str,
scope: str | None = None,
state: str | None = None,
prompt: str | None = None,
) -> dict[str, str]:
"""创建飞书用户 OAuth 授权 URL；需要用户态访问时把该链接发给用户。"""
scopes = scope.split() if scope else None
return {
"url": get_client().oauth.authorize_url(
redirect_uri,
scope=scopes,
state=state,
prompt=prompt,
)
}
@mcp.tool()
async def feishu_read_document_raw_content(
token: str,
doc_type: str | None = None,
user_access_token: str | None = None,
lang: int | None = 1,
) -> dict[str, str | None]:
"""
使用请求用户的 token 读取飞书文档纯文本。
用户私有文档必须使用 `user_access_token`，不要用机器人租户 token 读取。
"""
token = _required(token, "token")
user_access_token = _required_user_access_token(user_access_token, "read document content")
reference = DocumentReference(token=token, doc_type=doc_type)
content = await raw_document_content(get_client(user_access_token), reference, lang=lang)
return {"token": token, "doc_type": doc_type, "content": content}
@mcp.tool()
async def feishu_search_wiki(
query: str,
user_access_token: str | None = None,
space_id: str | None = None,
max_items: int | None = 10,
) -> dict[str, Any]:
"""
搜索请求用户可见的飞书知识库节点。
需要 `user_access_token`，确保结果受用户自身权限约束。
"""
query = _required(query, "query")
user_access_token = _required_user_access_token(user_access_token, "search user-visible documents")
items = await get_client(user_access_token).wiki.search(query, space_id=space_id, max_items=max_items)
return {"items": _plain(items)}
@mcp.tool()
async def feishu_list_drive_files(
user_access_token: str | None = None,
folder_token: str | None = None,
max_items: int | None = 20,
) -> dict[str, Any]:
"""
列出请求用户可见的云空间文件。
需要 `user_access_token`，确保结果受用户自身权限约束。
"""
user_access_token = _required_user_access_token(user_access_token, "list user-visible files")
items = await get_client(user_access_token).drive.files.list(folder_token=folder_token, max_items=max_items)
return {"items": _plain(items)}
@mcp.tool()
def feishu_pdf_to_text(pdf_base64: str, max_chars: int | None = PDF_DEFAULT_MAX_CHARS) -> dict[str, Any]:
"""
从 base64 编码的 PDF 中提取文本。
输入可以是裸 base64，也可以是 `data:application/pdf;base64` URL。
"""
return _pdf_to_text(_decode_base64_data(pdf_base64), max_chars=max_chars)
@mcp.tool()
def feishu_pdf_to_images(
pdf_base64: str,
start_page: int = 1,
max_pages: int = PDF_DEFAULT_MAX_PAGES,
zoom: float = PDF_DEFAULT_ZOOM,
) -> dict[str, Any]:
"""
将 base64 编码 PDF 的指定页面渲染为 PNG data URL。
页码从 1 开始；当首批页面不足以检查 PDF 时使用。
"""
return _pdf_to_images(
_decode_base64_data(pdf_base64),
start_page=start_page,
max_pages=max_pages,
zoom=zoom,
)
@mcp.tool()
async def feishu_message_pdf_to_text(
message_id: str,
file_key: str,
user_access_token: str | None = None,
resource_type: str = "file",
max_chars: int | None = PDF_DEFAULT_MAX_CHARS,
) -> dict[str, Any]:
"""下载飞书消息中的 PDF 资源并提取文本。"""
user_access_token = _required_user_access_token(user_access_token, "read message resources")
data = await get_client(user_access_token).im.get_resource(
_required(message_id, "message_id"),
_required(file_key, "file_key"),
resource_type=resource_type,
)
return _pdf_to_text(data, max_chars=max_chars)
@mcp.tool()
async def feishu_message_pdf_to_images(
message_id: str,
file_key: str,
user_access_token: str | None = None,
resource_type: str = "file",
start_page: int = 1,
max_pages: int = PDF_DEFAULT_MAX_PAGES,
zoom: float = PDF_DEFAULT_ZOOM,
) -> dict[str, Any]:
"""下载飞书消息中的 PDF 资源并把指定页面渲染为 PNG data URL。"""
user_access_token = _required_user_access_token(user_access_token, "read message resources")
data = await get_client(user_access_token).im.get_resource(
_required(message_id, "message_id"),
_required(file_key, "file_key"),
resource_type=resource_type,
)
return _pdf_to_images(data, start_page=start_page, max_pages=max_pages, zoom=zoom)
@mcp.tool()
async def feishu_query_freebusy(
time_min: str,
time_max: str,
user_id: str | None = None,
room_id: str | None = None,
user_access_token: str | None = None,
timezone: str = DEFAULT_TIMEZONE,
) -> dict[str, Any]:
"""查询用户或会议室的飞书日历忙闲状态。"""
user_access_token = _required_user_access_token(user_access_token, "query calendar free/busy")
body = freebusy_body(
time_min=time_min,
time_max=time_max,
user_id=user_id,
room_id=room_id,
timezone=timezone,
)
return await get_client(user_access_token).calendar.freebusy.query(body)
@mcp.tool()
async def feishu_get_primary_calendar(user_access_token: str | None = None) -> dict[str, Any]:
"""获取当前飞书身份的主日历。"""
user_access_token = _required_user_access_token(user_access_token, "get a primary calendar")
return await get_client(user_access_token).calendar.calendars.primary()
@mcp.tool()
async def feishu_list_calendar_events(
calendar_id: str,
start_time: str | None = None,
end_time: str | None = None,
user_access_token: str | None = None,
max_items: int | None = 20,
timezone: str = DEFAULT_TIMEZONE,
) -> list[NestedDict]:
"""列出指定飞书日历中的日程。"""
user_access_token = _required_user_access_token(user_access_token, "list calendar events")
return await get_client(user_access_token).calendar.events.list(
_required(calendar_id, "calendar_id"),
start_time=str(unix_seconds(start_time, timezone=timezone)) if start_time else None,
end_time=str(unix_seconds(end_time, timezone=timezone)) if end_time else None,
max_items=max_items,
)
@mcp.tool()
async def feishu_get_vc_meeting(
meeting_id: str,
user_access_token: str | None = None,
with_participants: bool | None = True,
with_meeting_ability: bool | None = None,
user_id_type: str | None = None,
) -> dict[str, Any]:
"""
获取请求用户可见的飞书视频会议详情。
需要 `user_access_token`，确保私有会议数据受用户自身权限约束。
"""
user_access_token = _required_user_access_token(user_access_token, "read meeting details")
return await get_client(user_access_token).vc.meetings.get(
_required(meeting_id, "meeting_id"),
with_participants=with_participants,
with_meeting_ability=with_meeting_ability,
user_id_type=user_id_type,
)
@mcp.tool()
async def feishu_list_vc_meetings_by_no(
meeting_no: str,
start_time: str,
end_time: str,
user_access_token: str | None = None,
max_items: int | None = 10,
timezone: str = DEFAULT_TIMEZONE,
) -> dict[str, Any]:
"""
按会议号和时间窗口列出飞书视频会议实例。
需要 `user_access_token`，确保私有会议数据受用户自身权限约束。
"""
user_access_token = _required_user_access_token(user_access_token, "list meeting records")
items = await get_client(user_access_token).vc.meetings.list_by_no(
_required(meeting_no, "meeting_no"),
str(unix_seconds(start_time, timezone=timezone)),
str(unix_seconds(end_time, timezone=timezone)),
max_items=max_items,
)
return {"items": _plain(items)}
@mcp.tool()
async def feishu_list_meeting_rooms(
building_id: str | None = None,
query: str | None = None,
min_capacity: int | None = None,
available_start_time: str | None = None,
available_end_time: str | None = None,
max_items: int | None = 20,
timezone: str = DEFAULT_TIMEZONE,
) -> dict[str, Any]:
"""列出或搜索飞书会议室，可按容量与可用性过滤。"""
if building_id:
rooms = await get_client().meeting_room.list(
building_id=building_id,
max_items=max_items,
)
else:
rooms = []
for building in await get_client().meeting_room.list_buildings(max_items=10):
current_building_id = building.get("building_id") or building.get("id")
if not current_building_id:
continue
current = await get_client().meeting_room.list(
building_id=current_building_id,
max_items=max_items,
)
rooms.extend(current)
if max_items is not None and len(rooms) >= max_items:
rooms = rooms[:max_items]
break
rooms = _filter_rooms(rooms, query=query, min_capacity=min_capacity)
if available_start_time and available_end_time and rooms:
freebusy = await get_client().meeting_room.freebusy(
[room["room_id"] for room in rooms if room.get("room_id")],
time_min=available_start_time,
time_max=available_end_time,
timezone=timezone,
)
rooms = _available_rooms(rooms, freebusy)
return {"items": rooms}
@mcp.tool()
async def feishu_get_meeting_rooms(room_ids: list[str]) -> dict[str, Any]:
"""按 `room_id` 获取飞书会议室详情。"""
return {"items": await get_client().meeting_room.batch_get(room_ids)}
@mcp.tool()
async def feishu_list_meeting_room_buildings(max_items: int | None = 20) -> dict[str, Any]:
"""列出飞书会议室建筑。"""
return {"items": await get_client().meeting_room.list_buildings(max_items=max_items)}
@mcp.tool()
async def feishu_query_meeting_room_freebusy(
room_ids: list[str],
time_min: str,
time_max: str,
timezone: str = DEFAULT_TIMEZONE,
) -> dict[str, Any]:
"""按 `room_id` 查询飞书会议室忙闲状态。"""
return await get_client().meeting_room.freebusy(
room_ids,
time_min=time_min,
time_max=time_max,
timezone=timezone,
)
@mcp.tool()
async def feishu_get_approval_definition(
approval_code: str,
locale: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
"""获取飞书审批定义，包括用于字段映射的表单元数据。"""
return await get_client().approval.definitions.get(
_required(approval_code, "approval_code"),
locale=locale,
user_id=user_id,
)
@mcp.tool()
async def feishu_create_calendar_event(
calendar_id: str,
event: dict[str, Any] | None = None,
summary: str | None = None,
start_time: str | None = None,
end_time: str | None = None,
attendees: list[dict[str, Any]] | None = None,
user_access_token: str | None = None,
idempotency_key: str | None = None,
timezone: str = DEFAULT_TIMEZONE,
confirmed: bool = False,
) -> dict[str, Any]:
"""
创建飞书日程，并可选添加参与人或会议室。
调用该写工具前，调用方应先取得用户明确确认。
"""
_require_confirmed(confirmed)
calendar_id = _required(calendar_id, "calendar_id")
event_payload = event
if event_payload is None:
event_payload = calendar_event(
summary=_required(summary, "summary"),
start_time=_required(start_time, "start_time"),
end_time=_required(end_time, "end_time"),
timezone=timezone,
)
result = await get_client(user_access_token).calendar.events.create(
calendar_id,
event_payload,
idempotency_key=idempotency_key,
)
event_data = result.get("event") or {}
event_id = event_data.get("event_id")
if attendees and event_id:
attendee_result = await get_client(user_access_token).calendar.attendees.add(
calendar_id,
str(event_id),
calendar_attendees(attendees),
)
result["attendees_result"] = attendee_result
return result
@mcp.tool()
async def feishu_create_approval_instance(
approval_code: str,
form: dict[str, Any] | list[dict[str, Any]] | str,
department_id: str | None = None,
user_access_token: str | None = None,
confirmed: bool = False,
) -> dict[str, Any]:
"""
为请求用户创建飞书审批实例。
调用该写工具前，调用方应先取得用户明确确认。申请人从 `user_access_token` 解析；
不接受调用方提供的 `user_id` / `open_id` 字段。
"""
_require_confirmed(confirmed)
user_access_token = _required_user_access_token(user_access_token, "create approval instances")
user = await get_client().oauth.user_info(user_access_token)
user_id = user.get("user_id")
open_id = user.get("open_id")
if not user_id and not open_id:
raise ValueError("user_access_token did not resolve to an approval applicant id")
payload = approval_instance(
_required(approval_code, "approval_code"),
form=form,
user_id=user_id,
open_id=open_id,
department_id=department_id,
)
return await get_client().approval.instances.create(payload)
@mcp.tool()
async def feishu_create_bitable_record(
app_token: str,
table_id: str,
fields: dict[str, Any],
user_access_token: str | None = None,
confirmed: bool = False,
) -> dict[str, Any]:
"""
创建一条飞书多维表格记录。
调用该写工具前，调用方应先取得用户明确确认。
"""
_require_confirmed(confirmed)
record = bitable_record(fields)
return await get_client(user_access_token).bitable.records.create(
_required(app_token, "app_token"),
_required(table_id, "table_id"),
record.fields,
)
return mcp