"""Feishu OAuth and Channel API routes."""
import asyncio
import time
import uuid
from collections.abc import Awaitable, Callable
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from loguru import logger
from sqlalchemy import select, or_
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.permissions import check_agent_access, is_agent_creator, is_agent_expired
from app.core.security import get_current_user
from app.database import get_db
from app.models.channel_config import ChannelConfig
from app.models.user import User
from app.models.identity import IdentityProvider
from app.schemas.schemas import ChannelConfigCreate, ChannelConfigOut, TokenResponse, UserOut
from app.services.feishu_service import feishu_service
router = APIRouter(tags=["feishu"])
# Default LLM timeout for Feishu channel (fallback when model has no request_timeout set).
# The per-model request_timeout field takes precedence — see _get_llm_timeout().
_LLM_TIMEOUT_SECONDS_DEFAULT = 180.0
# Number of tool status lines to keep visible in the Feishu card.
# Shows the last N non-running lines plus any active "running" entry.
_TOOL_STATUS_KEEP_LINES = 20
def _get_llm_timeout(model) -> float:
"""Get effective LLM timeout for the Feishu channel.
Prefer the model-level request_timeout so each model can have its own
budget (local vLLM may need 300 s, cloud APIs often need only 60 s).
Falls back to _LLM_TIMEOUT_SECONDS_DEFAULT when the field is absent or zero.
"""
timeout = getattr(model, "request_timeout", None)
if timeout and float(timeout) > 0:
return float(timeout)
return _LLM_TIMEOUT_SECONDS_DEFAULT
class _SerialPatchQueue:
"""Serialize patch requests for one Feishu message to prevent out-of-order overwrite."""
def __init__(self):
self._tail: asyncio.Task | None = None
def enqueue(self, job_factory: Callable[[], Awaitable[None]]) -> None:
prev = self._tail
async def _runner():
if prev:
try:
await prev
except Exception as e:
logger.warning(f"[Feishu] Previous patch job failed before next job: {e}")
await job_factory()
self._tail = asyncio.create_task(_runner())
async def drain(self) -> None:
if self._tail:
await self._tail
# ─── OAuth ──────────────────────────────────────────────
from fastapi.responses import HTMLResponse, Response
@router.get("/auth/feishu/callback")
@router.post("/auth/feishu/callback", response_model=TokenResponse)
async def feishu_oauth_callback(
code: str,
state: str = None,
db: AsyncSession = Depends(get_db)
):
"""Handle Feishu OAuth callback — exchange code for user session."""
# Parse state if it's a UUID (session ID) or other context
from app.models.identity import SSOScanSession
tenant_id = None
if state:
try:
sid = uuid.UUID(state)
s_res = await db.execute(select(SSOScanSession).where(SSOScanSession.id == sid))
session = s_res.scalar_one_or_none()
if session:
tenant_id = session.tenant_id
except (ValueError, AttributeError):
pass
try:
# Use FeishuAuthProvider instead of legacy feishu_service
from app.services.auth_provider import FeishuAuthProvider
from app.models.identity import IdentityProvider
from app.config import get_settings
# Get Feishu credentials from settings
settings = get_settings()
feishu_config = {
"app_id": settings.FEISHU_APP_ID,
"app_secret": settings.FEISHU_APP_SECRET,
}
# Get or create provider via auth provider
provider = None
if tenant_id:
result = await db.execute(
select(IdentityProvider).where(
IdentityProvider.provider_type == "feishu",
IdentityProvider.tenant_id == tenant_id
)
)
provider = result.scalar_one_or_none()
auth_provider = FeishuAuthProvider(provider=provider, config=feishu_config)
# Ensure provider exists (will create if not)
await auth_provider._ensure_provider(db, tenant_id)
provider = auth_provider.provider
# Exchange code for user info
token_data = await auth_provider.exchange_code_for_token(code)
access_token = token_data.get("access_token", "")
user_info = await auth_provider.get_user_info(access_token)
# Find or create user
user, is_new = await auth_provider.find_or_create_user(db, user_info, tenant_id=tenant_id)
# Generate JWT token
from app.core.security import create_access_token
token = create_access_token(str(user.id), user.role)
except Exception as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Feishu auth failed: {e}")
# If this is an SSO session, store result and redirect to frontend completion
if state:
try:
sid = uuid.UUID(state)
s_res = await db.execute(select(SSOScanSession).where(SSOScanSession.id == sid))
session = s_res.scalar_one_or_none()
if session:
session.status = "authorized"
session.provider_type = "feishu"
session.user_id = user.id
session.access_token = token
session.error_msg = None
await db.commit()
return HTMLResponse(
f"""
SSO login successful. Redirecting...
"""
)
except Exception as e:
logger.exception("Failed to update SSO session (feishu) %s", e)
return TokenResponse(access_token=token, user=UserOut.model_validate(user))
# ─── Channel Config (per-agent Feishu bot) ──────────────
@router.post("/agents/{agent_id}/channel", response_model=ChannelConfigOut, status_code=status.HTTP_201_CREATED)
async def configure_channel(
agent_id: uuid.UUID,
data: ChannelConfigCreate,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
"""Configure Feishu bot credentials for a digital employee (wizard step 5)."""
agent, _access = await check_agent_access(db, current_user, agent_id)
if not is_agent_creator(current_user, agent):
raise HTTPException(status_code=403, detail="Only creator can configure channel")
# Check existing
result = await db.execute(select(ChannelConfig).where(
ChannelConfig.agent_id == agent_id,
ChannelConfig.channel_type == "feishu",
))
existing = result.scalar_one_or_none()
if existing:
existing.app_id = data.app_id
existing.app_secret = data.app_secret
existing.encrypt_key = data.encrypt_key
existing.verification_token = data.verification_token
existing.extra_config = data.extra_config or {}
existing.is_configured = True
await db.flush()
# Start/Stop WS client in background
from app.services.feishu_ws import feishu_ws_manager
import asyncio
mode = existing.extra_config.get("connection_mode", "webhook")
if mode == "websocket":
asyncio.create_task(feishu_ws_manager.start_client(agent_id, existing.app_id, existing.app_secret))
else:
asyncio.create_task(feishu_ws_manager.stop_client(agent_id))
return ChannelConfigOut.model_validate(existing)
config = ChannelConfig(
agent_id=agent_id,
channel_type=data.channel_type,
app_id=data.app_id,
app_secret=data.app_secret,
encrypt_key=data.encrypt_key,
verification_token=data.verification_token,
extra_config=data.extra_config or {},
is_configured=True,
)
db.add(config)
await db.flush()
# Start WS client in background
from app.services.feishu_ws import feishu_ws_manager
import asyncio
mode = config.extra_config.get("connection_mode", "webhook")
if mode == "websocket":
asyncio.create_task(feishu_ws_manager.start_client(agent_id, config.app_id, config.app_secret))
return ChannelConfigOut.model_validate(config)
@router.get("/agents/{agent_id}/channel", response_model=ChannelConfigOut)
async def get_channel_config(
agent_id: uuid.UUID,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
"""Get Feishu channel configuration for an agent."""
await check_agent_access(db, current_user, agent_id)
result = await db.execute(select(ChannelConfig).where(
ChannelConfig.agent_id == agent_id,
ChannelConfig.channel_type == "feishu",
))
config = result.scalar_one_or_none()
if not config:
raise HTTPException(status_code=404, detail="Channel not configured")
return ChannelConfigOut.model_validate(config)
@router.get("/agents/{agent_id}/channel/webhook-url")
async def get_webhook_url(agent_id: uuid.UUID, request: Request, db: AsyncSession = Depends(get_db)):
"""Get the webhook URL for this agent's Feishu bot."""
from app.services.platform_service import platform_service
public_base = await platform_service.get_public_base_url(db, request)
return {"webhook_url": f"{public_base}/api/channel/feishu/{agent_id}/webhook"}
@router.delete("/agents/{agent_id}/channel", status_code=status.HTTP_204_NO_CONTENT)
async def delete_channel_config(
agent_id: uuid.UUID,
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
"""Remove Feishu bot configuration for an agent."""
agent, _access = await check_agent_access(db, current_user, agent_id)
if not is_agent_creator(current_user, agent):
raise HTTPException(status_code=403, detail="Only creator can remove channel")
result = await db.execute(select(ChannelConfig).where(
ChannelConfig.agent_id == agent_id,
ChannelConfig.channel_type == "feishu",
))
config = result.scalar_one_or_none()
if not config:
raise HTTPException(status_code=404, detail="Channel not configured")
await db.delete(config)
# ─── Feishu Event Webhook ───────────────────────────────
# Simple in-memory dedup to avoid processing retried events
_processed_events: set[str] = set()
@router.post("/channel/feishu/{agent_id}/webhook")
async def feishu_event_webhook(
agent_id: uuid.UUID,
request: Request,
db: AsyncSession = Depends(get_db),
):
"""Handle Feishu event callback for a specific agent's bot."""
body = await request.json()
# Handle verification challenge
if "challenge" in body:
return {"challenge": body["challenge"]}
return await process_feishu_event(agent_id, body, db)
async def process_feishu_event(agent_id: uuid.UUID, body: dict, db: AsyncSession):
"""Core logic to process feishu events from both webhook and WS client."""
import json as _json
logger.info(f"[Feishu] Event processing for {agent_id}: event_type={body.get('header', {}).get('event_type', 'N/A')}")
# Deduplicate — Feishu retries on slow responses
# Only mark as processed AFTER successful handling so retries work on crash
event_id = body.get("header", {}).get("event_id", "")
if event_id in _processed_events:
return {"code": 0, "msg": "already processed"}
# Get channel config — filter by feishu since an agent can have multiple channels
result = await db.execute(
select(ChannelConfig).where(
ChannelConfig.agent_id == agent_id,
ChannelConfig.channel_type == "feishu",
)
)
config = result.scalar_one_or_none()
if not config:
return {"code": 1, "msg": "Channel not found"}
# Mark event as processed after config is loaded successfully
if event_id:
_processed_events.add(event_id)
# Keep set bounded
if len(_processed_events) > 1000:
_processed_events.clear()
# Handle events
event = body.get("event", {})
event_type = body.get("header", {}).get("event_type", "")
if event_type == "im.message.receive_v1":
message = event.get("message", {})
sender = event.get("sender", {}).get("sender_id", {})
sender_open_id = sender.get("open_id", "")
sender_user_id_from_event = sender.get("user_id", "") # tenant-stable ID, available directly in event body
msg_type = message.get("message_type", "text")
chat_type = message.get("chat_type", "p2p") # p2p or group
chat_id = message.get("chat_id", "")
logger.info(f"[Feishu] Received {msg_type} message, chat_type={chat_type}, from={sender_open_id}")
# ── Normalize post (rich text) → extract text + schedule image downloads ──
if msg_type == "post":
import json as _json_post
_post_body = _json_post.loads(message.get("content", "{}"))
# Feishu post content: {"title": "...", "content": [[{"tag":"text","text":"..."},...],...]}
# The content may be nested under a locale key like "zh_cn"
_paragraphs = _post_body.get("content", [])
if not _paragraphs:
# Try locale keys (zh_cn, en_us, etc.)
for _locale_key, _locale_val in _post_body.items():
if isinstance(_locale_val, dict) and "content" in _locale_val:
_paragraphs = _locale_val["content"]
break
_text_parts = []
_post_image_keys = []
for _para in _paragraphs:
_line_parts = []
for _elem in _para:
_tag = _elem.get("tag")
if _tag == "text":
_line_parts.append(_elem.get("text", ""))
elif _tag == "a":
_href = _elem.get("href", "")
_link_text = _elem.get("text", "")
_line_parts.append(f"{_link_text} ({_href})" if _href else _link_text)
elif _tag == "img":
_ik = _elem.get("image_key", "")
if _ik:
_post_image_keys.append(_ik)
if _line_parts:
_text_parts.append("".join(_line_parts))
_extracted_text = "\n".join(_text_parts).strip()
# Download images and embed as base64 for vision-capable models
_image_markers = []
if _post_image_keys:
import base64 as _b64
_msg_id = message.get("message_id", "")
from pathlib import Path as _PostPath
from app.config import get_settings as _post_gs
_post_settings = _post_gs()
_upload_dir = _PostPath(_post_settings.AGENT_DATA_DIR) / str(agent_id) / "workspace" / "uploads"
_upload_dir.mkdir(parents=True, exist_ok=True)
for _ik in _post_image_keys:
try:
_img_bytes = await feishu_service.download_message_resource(
config.app_id, config.app_secret, _msg_id, _ik, "image"
)
# Save to workspace
_save_path = _upload_dir / f"image_{_ik[-8:]}.jpg"
_save_path.write_bytes(_img_bytes)
logger.info(f"[Feishu] Saved post image to {_save_path} ({len(_img_bytes)} bytes)")
# Embed as base64 marker for vision models
_b64_data = _b64.b64encode(_img_bytes).decode("ascii")
_image_markers.append(f"[image_data:data:image/jpeg;base64,{_b64_data}]")
except Exception as _dl_err:
logger.error(f"[Feishu] Failed to download post image {_ik}: {_dl_err}")
# Build final text with embedded images
if not _extracted_text and _image_markers:
_extracted_text = "[用户发送了图片，请看图片内容]"
_final_content = _extracted_text
if _image_markers:
_final_content += "\n" + "\n".join(_image_markers)
# Rewrite as text message so existing handler processes it
message["content"] = _json_post.dumps({"text": _final_content})
msg_type = "text"
logger.info(f"[Feishu] Normalized post → text='{_extracted_text[:100]}', images={len(_image_markers)}")
if msg_type in ("file", "image"):
import asyncio as _asyncio
_asyncio.create_task(_handle_feishu_file(db, agent_id, config, message, sender_open_id, chat_type, chat_id))
return {"code": 0, "msg": "ok"}
if msg_type == "text":
import json
import re
content = json.loads(message.get("content", "{}"))
user_text = content.get("text", "")
# Strip @mention tags (e.g. @_user_1) from group messages
user_text = re.sub(r'@_user_\d+', '', user_text).strip()
if not user_text:
return {"code": 0, "msg": "empty message after stripping mentions"}
# Detect task creation intent
task_match = re.search(
r'(?:创建|新建|添加|建一个|帮我建)(?:一个)?(?:任务|待办|todo)[，,：:\s]*(.+)',
user_text, re.IGNORECASE
)
# Determine conversation_id for history isolation
# Group chats: use chat_id; P2P chats: prefer user_id (tenant-stable)
if chat_type == "group" and chat_id:
conv_id = f"feishu_group_{chat_id}"
else:
conv_id = f"feishu_p2p_{sender_user_id_from_event or sender_open_id}"
# Load recent conversation history via session (session UUID may already exist)
from app.models.audit import ChatMessage
from app.models.agent import Agent as AgentModel
from app.services.channel_session import find_or_create_channel_session
agent_r = await db.execute(select(AgentModel).where(AgentModel.id == agent_id))
agent_obj = agent_r.scalar_one_or_none()
creator_id = agent_obj.creator_id if agent_obj else agent_id
from app.models.agent import DEFAULT_CONTEXT_WINDOW_SIZE
ctx_size = (agent_obj.context_window_size or DEFAULT_CONTEXT_WINDOW_SIZE) if agent_obj else DEFAULT_CONTEXT_WINDOW_SIZE
# Pre-resolve session so history lookup uses the UUID (session created later if new)
_pre_sess_r = await db.execute(
select(__import__('app.models.chat_session', fromlist=['ChatSession']).ChatSession).where(
__import__('app.models.chat_session', fromlist=['ChatSession']).ChatSession.agent_id == agent_id,
__import__('app.models.chat_session', fromlist=['ChatSession']).ChatSession.external_conv_id == conv_id,
)
)
_pre_sess = _pre_sess_r.scalar_one_or_none()
_history_conv_id = str(_pre_sess.id) if _pre_sess else conv_id
history_result = await db.execute(
select(ChatMessage)
.where(ChatMessage.agent_id == agent_id, ChatMessage.conversation_id == _history_conv_id)
.order_by(ChatMessage.created_at.desc())
.limit(ctx_size)
)
history_msgs = history_result.scalars().all()
history = [{"role": m.role, "content": m.content} for m in reversed(history_msgs)]
# --- Resolve Feishu sender identity & find/create platform user ---
import uuid as _uuid
import httpx as _httpx
sender_name = ""
sender_user_id_feishu = sender_user_id_from_event # tenant-level user_id, pre-filled from event body
extra_info: dict | None = None
try:
async with _httpx.AsyncClient() as _client:
_tok_resp = await _client.post(
"https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal",
json={"app_id": config.app_id, "app_secret": config.app_secret},
)
_app_token = _tok_resp.json().get("app_access_token", "")
if _app_token:
_user_resp = await _client.get(
f"https://open.feishu.cn/open-apis/contact/v3/users/{sender_open_id}",
params={"user_id_type": "open_id"},
headers={"Authorization": f"Bearer {_app_token}"},
)
_user_data = _user_resp.json()
logger.info(f"[Feishu] Sender resolve: code={_user_data.get('code')}, msg={_user_data.get('msg', '')}")
if _user_data.get("code") == 0:
_user_info = _user_data.get("data", {}).get("user", {})
sender_name = _user_info.get("name", "")
sender_user_id_feishu = _user_info.get("user_id", "")
sender_email = _user_info.get("email", "") or _user_info.get("enterprise_email", "")
# Feishu contact API returns 'avatar' as a dict
# (keys: avatar_240, avatar_640, avatar_origin), NOT a plain URL.
# We must extract a string to avoid a DataError when writing to the DB.
_raw_avatar = _user_info.get("avatar")
if isinstance(_raw_avatar, dict):
_avatar_url = (
_raw_avatar.get("avatar_240")
or _raw_avatar.get("avatar_640")
or _raw_avatar.get("avatar_origin")
or ""
)
else:
_avatar_url = _raw_avatar or ""
extra_info = {
"name": sender_name,
"email": sender_email,
"mobile": _user_info.get("mobile"),
"avatar_url": _avatar_url,
"unionid": _user_info.get("user_id"), # tenant-level user_id
"open_id": sender_open_id,
}
logger.info(f"[Feishu] Resolved sender: {sender_name} (user_id={sender_user_id_feishu})")
# Cache sender info so feishu_user_search can find them by name
if sender_name and sender_open_id:
try:
import pathlib as _pl, json as _cj, time as _ct
_safe_id = str(agent_id).replace("..", "").replace("/", "")
_cache = _pl.Path(f"/data/workspaces/{_safe_id}/feishu_contacts_cache.json")
_cache.parent.mkdir(parents=True, exist_ok=True)
_existing = {}
if _cache.exists():
try:
_existing = _cj.loads(_cache.read_text())
except Exception:
pass
# Key by user_id when available (tenant-stable), fallback to open_id
_users = {}
for _u in _existing.get("users", []):
_key = _u.get("user_id") or _u.get("open_id", "")
_users[_key] = _u
_cache_key = sender_user_id_feishu or sender_open_id
_users[_cache_key] = {
"open_id": sender_open_id,
"name": sender_name,
"email": sender_email,
"user_id": sender_user_id_feishu,
}
_cache.write_text(_cj.dumps(
{"ts": _ct.time(), "users": list(_users.values())},
ensure_ascii=False,
), encoding="utf-8")
import os as _os
_os.chmod(str(_cache), 0o600)
except Exception as _ce:
logger.error(f"[Feishu] Cache write failed: {_ce}")
except Exception as e:
logger.error(f"[Feishu] Failed to resolve sender: {e}")
# Resolve channel user via unified service (uses OrgMember + SSO patterns)
from app.services.channel_user_service import channel_user_service
platform_user = await channel_user_service.resolve_channel_user(
db=db,
agent=agent_obj,
channel_type="feishu",
external_user_id=sender_open_id,
extra_info=extra_info,
)
platform_user_id = platform_user.id
# ── Find-or-create a ChatSession via external_conv_id (DB-based, no cache needed) ──
from datetime import datetime as _dt, timezone as _tz
_is_group = (chat_type == "group")
_sess = await find_or_create_channel_session(
db=db,
agent_id=agent_id,
user_id=platform_user_id if not _is_group else creator_id,
external_conv_id=conv_id,
source_channel="feishu",
first_message_title=user_text,
is_group=_is_group,
group_name=f"Feishu Group {chat_id[:8]}" if _is_group else None,
)
session_conv_id = str(_sess.id)
# Save user message
db.add(ChatMessage(agent_id=agent_id, user_id=platform_user_id, role="user", content=user_text, conversation_id=session_conv_id))
_sess.last_message_at = _dt.now(_tz.utc)
await db.commit()
# Prepend sender identity so the agent knows who is talking
llm_user_text = user_text
if sender_name:
id_part = f" (ID: {sender_user_id_feishu})" if sender_user_id_feishu else ""
llm_user_text = f"[发送者: {sender_name}{id_part}] {user_text}"
# ── Inject recent uploaded file context ──────────────────────────
# Check the uploads directory for recently modified files (within 30 min).
# This is more reliable than scanning DB history, because the file save
# to disk always succeeds even if the DB transaction fails.
try:
import time as _time
import pathlib as _pl
from app.config import get_settings as _gs
_upload_dir = _pl.Path(_gs().AGENT_DATA_DIR) / str(agent_id) / "workspace" / "uploads"
_recent_file_path = None
if _upload_dir.exists() and "uploads/" not in user_text and "workspace/" not in user_text:
_now = _time.time()
_candidates = sorted(
_upload_dir.iterdir(),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
for _fp in _candidates:
if _fp.is_file() and (_now - _fp.stat().st_mtime) < 1800: # 30 min
_recent_file_path = f"uploads/{_fp.name}"
break
if _recent_file_path:
# _recent_file_path is relative to uploads dir; agent workspace root is
# AGENT_DATA_DIR/{agent_id}/, so the correct relative path is workspace/uploads/
_ws_rel_path = f"workspace/{_recent_file_path}"
llm_user_text = (
llm_user_text
+ f"\n\n[系统提示：用户刚上传了文件，路径为工作区 `{_ws_rel_path}`。"
f"如果用户的指令涉及这篇文章、这个文件、这份文档等，"
f"请立即调用 read_document(path=\"{_ws_rel_path}\") 读取内容，不要先用 list_files 验证，直接读取即可。]"
)
logger.info(f"[Feishu] Injected recent file hint: {_ws_rel_path}")
except Exception as _fe:
logger.error(f"[Feishu] File injection error: {_fe}")
# Set sender open_id contextvar so calendar tool can auto-invite the requester
from app.services.agent_tools import channel_feishu_sender_open_id as _cfso
_cfso_token = _cfso.set(sender_open_id)
# Set channel_file_sender contextvar so the agent can send files back via Feishu
from app.services.agent_tools import channel_file_sender as _cfs
_reply_to_id = chat_id if chat_type == "group" else sender_open_id
_rid_type = "chat_id" if chat_type == "group" else "open_id"
async def _feishu_file_sender(file_path, msg: str = ""):
try:
await feishu_service.upload_and_send_file(
config.app_id, config.app_secret,
_reply_to_id, file_path,
receive_id_type=_rid_type,
accompany_msg=msg,
)
except Exception as _upload_err:
# Fallback: send a download link when upload permission is not granted
from pathlib import Path as _P
from app.config import get_settings as _gs_fallback
_fs = _gs_fallback()
_base_url = getattr(_fs, 'BASE_URL', '').rstrip('/') or ''
_fp = _P(file_path)
_ws_root = _P(_fs.AGENT_DATA_DIR)
try:
_rel = str(_fp.relative_to(_ws_root / str(agent_id)))
except ValueError:
_rel = _fp.name
_fallback_parts = []
if msg:
_fallback_parts.append(msg)
if _base_url:
_dl_url = f"{_base_url}/api/agents/{agent_id}/files/download?path={_rel}"
_fallback_parts.append(f"📎 {_fp.name}\n🔗 {_dl_url}")
_fallback_parts.append(
f"⚠️ 文件直接发送失败（{_upload_err}）\n"
"如需 Agent 直接发飞书文件，请在飞书开放平台为应用开启 "
"`im:resource`（即 `im:resource:upload`）权限并发布版本。"
)
await feishu_service.send_message(
config.app_id, config.app_secret,
_reply_to_id, "text",
_json.dumps({"text": "\n\n".join(_fallback_parts)}),
receive_id_type=_rid_type,
)
_cfs_token = _cfs.set(_feishu_file_sender)
# Set up streaming response via CardKit (primary) or IM patch (fallback)
import json as _json_card
cardkit_card_id: str | None = None
cardkit_sequence: int = 0
msg_id_for_patch: str | None = None
_reply_target = chat_id if chat_type == "group" and chat_id else sender_open_id
_rid_type = "chat_id" if chat_type == "group" and chat_id else "open_id"
init_card = {
"schema": "2.0",
"config": {
"streaming_mode": True,
"locales": ["zh_cn", "en_us"],
"summary": {"content": "思考中..."},
},
"body": {
"elements": [
{"tag": "markdown", "content": "", "text_align": "left", "text_size": "normal_v2", "element_id": "streaming_content"},
{"tag": "markdown", "content": " ", "icon": {"tag": "custom_icon", "img_key": "img_v3_02vb_496bec09-4b43-4773-ad6b-0cdd103cd2bg", "size": "16px 16px"}, "element_id": "loading_icon"},
]
},
}
try:
cardkit_card_id = await feishu_service.create_card_entity(
config.app_id, config.app_secret, init_card
)
cardkit_sequence = 1
await feishu_service.send_card_by_card_id(
config.app_id, config.app_secret, _reply_target, cardkit_card_id,
receive_id_type=_rid_type,
)
logger.info(f"[Feishu] CardKit card created and sent: card_id={cardkit_card_id}")
except Exception as e:
logger.warning(f"[Feishu] CardKit flow failed, falling back to IM patch: {e}")
cardkit_card_id = None
init_card_fallback = {
"config": {"update_multi": True},
"header": {"template": "blue", "title": {"content": "思考中...", "tag": "plain_text"}},
"elements": [{"tag": "markdown", "content": "..."}],
}
try:
init_resp = await feishu_service.send_message(
config.app_id, config.app_secret, _reply_target, "interactive",
_json_card.dumps(init_card_fallback), receive_id_type=_rid_type, stage="stream_init_card",
)
msg_id_for_patch = init_resp.get("data", {}).get("message_id")
except Exception as e2:
logger.error(f"[Feishu] Fallback init card also failed: {e2}")
_stream_buffer = []
_thinking_buffer = []
_last_flush_time = time.time()
_FLUSH_INTERVAL_CARDKIT = 0.5
_FLUSH_INTERVAL_PATCH = 1.0
_agent_name = agent_obj.name if agent_obj else "AI 回复"
_tool_status_running: dict[str, str] = {}
_tool_status_done: list[str] = []
_patch_queue = _SerialPatchQueue()
_heartbeat_task: asyncio.Task | None = None
_llm_done = False
_last_flushed_hash: int = 0
_last_flushed_text: str = ""
_flush_lock = asyncio.Lock()
def _build_card(
answer_text: str,
thinking_text: str = "",
streaming: bool = False,
tool_status_lines: list[str] | None = None,
agent_name: str | None = None,
) -> dict:
"""Build a Feishu interactive card for streaming replies.
Args:
answer_text: Main reply text (may be partial during streaming).
thinking_text: Reasoning/thinking content shown in a collapsed section.
streaming: If True, appends a cursor glyph to indicate in-progress output.
tool_status_lines: Override list for image streaming (which maintains its
own done-list; pass None to use the default text-streaming state).
agent_name: Override the default _agent_name (for image streaming context).
"""
_name = agent_name if agent_name is not None else _agent_name
elements = []
# Tool status section.
# For the primary text-streaming path we use the split running/done dicts;
# callers may pass an explicit list (image streaming) as override.
if tool_status_lines is not None:
# Caller-supplied override (image path): plain list, no split needed.
if tool_status_lines:
elements.append({
"tag": "markdown",
"content": "\n".join(tool_status_lines[-_TOOL_STATUS_KEEP_LINES:]),
})
elements.append({"tag": "hr"})
else:
# Primary text-streaming path: show done history + any still-running tools.
# _tool_status_running entries are removed when the tool completes,
# so only genuinely in-flight tools appear here.
done_visible = _tool_status_done[-_TOOL_STATUS_KEEP_LINES:]
running_visible = list(_tool_status_running.values())
all_visible = done_visible + running_visible
if all_visible:
elements.append({
"tag": "markdown",
"content": "\n".join(all_visible),
})
elements.append({"tag": "hr"})
# Thinking section: collapsed grey block
if thinking_text:
think_preview = thinking_text[:200].replace("\n", " ")
elements.append({
"tag": "markdown",
"content": f"💭 **Thinking**\n{think_preview}{'...' if len(thinking_text) > 200 else ''}",
})
elements.append({"tag": "hr"})
body = answer_text + ("▌" if streaming and answer_text else ("..." if streaming else ""))
elements.append({"tag": "markdown", "content": body or "..."})
return {
"config": {"update_multi": True},
"header": {
"template": "blue",
"title": {"content": _name, "tag": "plain_text"},
},
"elements": elements,
}
def _build_final_cardkit_card(answer_text: str, thinking_text: str = "") -> dict:
elements = []
if thinking_text:
elements.append({
"tag": "collapsible_panel",
"expanded": False,
"header": {
"title": {"tag": "markdown", "content": f"💭 Thinking... ({len(thinking_text)} chars)"},
"vertical_align": "center",
"icon": {"tag": "standard_icon", "token": "down-small-ccm_outlined", "size": "16px 16px"},
"icon_position": "follow_text",
"icon_expanded_angle": -180,
},
"border": {"color": "grey", "corner_radius": "5px"},
"elements": [{"tag": "markdown", "content": thinking_text, "text_size": "notation"}],
})
elements.append({"tag": "markdown", "content": answer_text or "..."})
return {
"schema": "2.0",
"config": {"wide_screen_mode": True, "update_multi": True},
"body": {"elements": elements},
}
async def _queue_patch_card(card: dict, stage: str) -> None:
if not msg_id_for_patch:
return
payload = _json_card.dumps(card)
async def _job():
try:
await feishu_service.patch_message(
config.app_id,
config.app_secret,
msg_id_for_patch,
payload,
stage=stage,
)
except Exception as e:
logger.warning(f"[Feishu] Patch failed (stage={stage}, message_id={msg_id_for_patch}): {e}")
_patch_queue.enqueue(_job)
async def _flush_stream(reason: str, force: bool = False):
nonlocal _last_flush_time, _last_flushed_hash, cardkit_sequence, _last_flushed_text
if not cardkit_card_id and not msg_id_for_patch:
return
async with _flush_lock:
logger.debug(f"[Feishu] flush({reason}): seq={cardkit_sequence}")
now = time.time()
flush_interval = _FLUSH_INTERVAL_CARDKIT if cardkit_card_id else _FLUSH_INTERVAL_PATCH
if not force and now - _last_flush_time < flush_interval:
return
accumulated = "".join(_stream_buffer)
if cardkit_card_id:
# Build composite content: tool status lines + answer text.
# This mirrors the IM Patch path where _build_card() includes the
# tool status section, so CardKit users also see which tools are
# running or completed during the LLM turn.
done_visible = _tool_status_done[-_TOOL_STATUS_KEEP_LINES:]
running_visible = list(_tool_status_running.values())
all_tool_lines = done_visible + running_visible
if all_tool_lines:
tool_section = "\n".join(all_tool_lines)
cardkit_text = f"{tool_section}\n---\n{accumulated}" if accumulated else tool_section
else:
cardkit_text = accumulated
if cardkit_text != _last_flushed_text:
cardkit_sequence += 1
try:
await asyncio.wait_for(
feishu_service.stream_card_content(
config.app_id, config.app_secret,
cardkit_card_id, "streaming_content",
cardkit_text, cardkit_sequence,
),
timeout=5.0,
)
_last_flushed_text = cardkit_text
except asyncio.TimeoutError:
logger.warning(f"[Feishu] CardKit stream timed out, seq={cardkit_sequence}")
except Exception as e:
logger.warning(f"[Feishu] CardKit stream failed: {e}")
elif msg_id_for_patch:
card = _build_card(accumulated, "".join(_thinking_buffer), streaming=True)
current_hash = hash(accumulated + "".join(_thinking_buffer) + str(_tool_status_done) + str(list(_tool_status_running.values())))
if reason == "heartbeat" and current_hash == _last_flushed_hash:
return
_last_flushed_hash = current_hash
await _queue_patch_card(card, stage=f"stream_{reason}")
_last_flush_time = now
async def _ws_on_chunk(text: str):
if not cardkit_card_id and not msg_id_for_patch:
return
_stream_buffer.append(text)
await _flush_stream("chunk")
async def _ws_on_thinking(text: str):
if not cardkit_card_id and not msg_id_for_patch:
return
_thinking_buffer.append(text)
await _flush_stream("thinking")
async def _ws_on_tool_call(evt: dict):
"""Receive tool call status events and update the card's progress section.
Uses the tool's call_id as the dict key so each tool shows only its
latest state. When a tool completes the "running" entry is removed from
_tool_status_running and a "done" line is appended to _tool_status_done,
ensuring finished tools never linger as ⏳ in the card.
"""
tool_name = evt.get("name") or "unknown_tool"
# Use call_id when available (unique per invocation); fall back to name.
call_id = evt.get("call_id") or tool_name
status = (evt.get("status") or "").lower()
if status == "running":
# Register as in-flight; will be removed when "done" arrives.
_tool_status_running[call_id] = f"⏳ Tool running: `{tool_name}`"
elif status == "done":
# Remove from running dict so the ⏳ icon disappears immediately.
_tool_status_running.pop(call_id, None)
_tool_status_done.append(f"✅ Tool done: `{tool_name}`")
else:
_tool_status_running.pop(call_id, None)
_tool_status_done.append(f"ℹ️ Tool update: `{tool_name}` ({status or 'unknown'})")
await _flush_stream("tool")
async def _heartbeat():
while not _llm_done:
await asyncio.sleep(_FLUSH_INTERVAL_CARDKIT if cardkit_card_id else _FLUSH_INTERVAL_PATCH)
await _flush_stream("heartbeat")
if cardkit_card_id or msg_id_for_patch:
_heartbeat_task = asyncio.create_task(_heartbeat())
# Call LLM with history and streaming callback
try:
reply_text = await _call_agent_llm(
db,
agent_id,
llm_user_text,
history=history,
user_id=platform_user_id,
on_chunk=_ws_on_chunk,
on_thinking=_ws_on_thinking,
on_tool_call=_ws_on_tool_call,
)
finally:
_llm_done = True
if _heartbeat_task:
_heartbeat_task.cancel()
try:
await _heartbeat_task
except (Exception, asyncio.CancelledError):
pass
_cfs.reset(_cfs_token)
_cfso.reset(_cfso_token)
logger.info(f"[Feishu] LLM reply: {reply_text[:100]}")
# Send final card update or fallback text
if cardkit_card_id:
try:
cardkit_sequence += 1
await asyncio.wait_for(
feishu_service.set_card_streaming_mode(
config.app_id, config.app_secret,
cardkit_card_id, 0, cardkit_sequence,
),
timeout=10.0,
)
cardkit_sequence += 1
final_card = _build_final_cardkit_card(reply_text, "".join(_thinking_buffer))
await asyncio.wait_for(
feishu_service.update_cardkit_card(
config.app_id, config.app_secret,
cardkit_card_id, final_card, cardkit_sequence,
),
timeout=10.0,
)
except Exception as e:
logger.error(f"[Feishu] CardKit final update failed: {e}")
try:
await feishu_service.send_message(
config.app_id, config.app_secret, _reply_target, "text",
_json.dumps({"text": reply_text}), receive_id_type=_rid_type,
stage="stream_final_fallback_text",
)
except Exception as e2:
logger.error(f"[Feishu] CardKit fallback text also failed: {e2}")
elif msg_id_for_patch:
try:
await _patch_queue.drain()
except Exception as e:
logger.warning(f"[Feishu] Drain patch queue failed before final patch: {e}")
final_card = _build_card(
reply_text,
"".join(_thinking_buffer),
streaming=False,
)
try:
await feishu_service.patch_message(
config.app_id,
config.app_secret,
msg_id_for_patch,
_json_card.dumps(final_card),
stage="stream_final",
)
except Exception as e:
logger.error(f"[Feishu] Final card patch failed: {e}")
try:
await feishu_service.send_message(
config.app_id, config.app_secret, _reply_target, "text",
_json.dumps({"text": reply_text}), receive_id_type=_rid_type,
stage="stream_final_fallback_text",
)
except Exception as e2:
logger.error(f"[Feishu] Fallback text also failed: {e2}")
else:
try:
await feishu_service.send_message(
config.app_id, config.app_secret, _reply_target, "text",
_json.dumps({"text": reply_text}), receive_id_type=_rid_type,
stage="stream_no_card_fallback_text",
)
except Exception as e:
logger.error(f"[Feishu] Failed to send fallback message: {e}")
# Log activity
from app.services.activity_logger import log_activity
await log_activity(agent_id, "chat_reply", f"回复了飞书消息: {reply_text[:80]}", detail={"channel": "feishu", "user_text": user_text[:200], "reply": reply_text[:500]})
# If task creation detected, create a real Task record
if task_match:
task_title = task_match.group(1).strip()
if task_title:
try:
from app.models.task import Task as TaskModel
from app.models.agent import Agent as AgentModel
from app.services.task_executor import execute_task
import asyncio as _asyncio
# Find the agent's creator to use as task creator
agent_r = await db.execute(select(AgentModel).where(AgentModel.id == agent_id))
agent_obj = agent_r.scalar_one_or_none()
creator_id = agent_obj.creator_id if agent_obj else agent_id
task_obj = TaskModel(
agent_id=agent_id,
title=task_title,
created_by=creator_id,
status="pending",
priority="medium",
)
db.add(task_obj)
await db.commit()
await db.refresh(task_obj)
_asyncio.create_task(execute_task(task_obj.id, agent_id))
reply_text += f"\n\n📋 已同步创建任务到任务面板：【{task_title}】"
logger.info(f"[Feishu] Created task: {task_title}")
except Exception as e:
logger.error(f"[Feishu] Failed to create task: {e}")
# Save assistant reply to history (use platform_user_id so messages stay in one session)
db.add(ChatMessage(agent_id=agent_id, user_id=platform_user_id, role="assistant", content=reply_text, conversation_id=session_conv_id))
_sess.last_message_at = _dt.now(_tz.utc)
await db.commit()
return {"code": 0, "msg": "ok"}
IMPORT_RE = None # lazy sentinel
_FILE_ACK_MESSAGES = [
"收到你的文件，请问有什么需要帮忙的？",
"文件收到了！你想让我怎么处理它？",
"好的，我已经收到这份文件，请告诉我你的需求~",
"已收到文件，随时准备好为你处理！",
"收到！请问希望我对这份文件做什么？",
]
async def _handle_feishu_file(db, agent_id, config, message, sender_open_id, chat_type, chat_id):
"""Handle incoming file or image messages from Feishu (runs as a background task)."""
import asyncio, random, json
from pathlib import Path
from app.config import get_settings
from app.models.audit import ChatMessage
from app.models.agent import Agent as AgentModel
from app.models.user import User as UserModel
from app.services.channel_session import find_or_create_channel_session
from app.core.security import hash_password
from app.database import async_session as _async_session
from datetime import datetime as _dt, timezone as _tz
import uuid as _uuid
from sqlalchemy import select as _select
msg_type = message.get("message_type", "file")
message_id = message.get("message_id", "")
content = json.loads(message.get("content", "{}"))
# Extract file key and name
if msg_type == "image":
file_key = content.get("image_key", "")
filename = f"image_{file_key[-8:]}.jpg" if file_key else "image.jpg"
res_type = "image"
else:
file_key = content.get("file_key", "")
filename = content.get("file_name") or f"file_{file_key[-8:]}.bin"
res_type = "file"
if not file_key:
logger.warning(f"[Feishu] No file_key in {msg_type} message")
return
# Resolve workspace upload dir
settings = get_settings()
upload_dir = Path(settings.AGENT_DATA_DIR) / str(agent_id) / "workspace" / "uploads"
upload_dir.mkdir(parents=True, exist_ok=True)
save_path = upload_dir / filename
# Download the file
try:
file_bytes = await feishu_service.download_message_resource(
config.app_id, config.app_secret, message_id, file_key, res_type
)
save_path.write_bytes(file_bytes)
logger.info(f"[Feishu] Saved {msg_type} to {save_path} ({len(file_bytes)} bytes)")
except Exception as e:
logger.error(f"[Feishu] Failed to download {msg_type}: {e}")
err_tip = "抱歉，文件下载失败。可能原因：机器人缺少 `im:resource` 权限（文件读取）。\n请在飞书开放平台 → 权限管理 → 批量导入权限 JSON → 重新发布机器人版本后重试。"
try:
import json as _j
if chat_type == "group" and chat_id:
await feishu_service.send_message(config.app_id, config.app_secret, chat_id, "text", _j.dumps({"text": err_tip}), receive_id_type="chat_id")
else:
await feishu_service.send_message(config.app_id, config.app_secret, sender_open_id, "text", _j.dumps({"text": err_tip}))
except Exception as e2:
logger.error(f"[Feishu] Also failed to send error tip: {e2}")
return
# Resolve platform user and session using a fresh db session
async with _async_session() as db:
agent_r = await db.execute(_select(AgentModel).where(AgentModel.id == agent_id))
agent_obj = agent_r.scalar_one_or_none()
# Resolve sender's Feishu user_id (more stable than open_id)
sender_user_id_feishu = ""
extra_info: dict | None = None
try:
import httpx as _hx
async with _hx.AsyncClient() as _fc:
_tr = await _fc.post(
"https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal",
json={"app_id": config.app_id, "app_secret": config.app_secret},
)
_at = _tr.json().get("app_access_token", "")
if _at:
_ur = await _fc.get(
f"https://open.feishu.cn/open-apis/contact/v3/users/{sender_open_id}",
params={"user_id_type": "open_id"},
headers={"Authorization": f"Bearer {_at}"},
)
_ud = _ur.json()
if _ud.get("code") == 0:
_user_info = _ud.get("data", {}).get("user", {})
sender_user_id_feishu = _user_info.get("user_id", "")
# Feishu contact API returns 'avatar' as a dict
# (keys: avatar_240, avatar_640, avatar_origin), NOT a plain URL.
_raw_avatar = _user_info.get("avatar")
if isinstance(_raw_avatar, dict):
_avatar_url = (
_raw_avatar.get("avatar_240")
or _raw_avatar.get("avatar_640")
or _raw_avatar.get("avatar_origin")
or ""
)
else:
_avatar_url = _raw_avatar or ""
extra_info = {
"name": _user_info.get("name"),
"avatar_url": _avatar_url,
"email": _user_info.get("email"),
"mobile": _user_info.get("mobile"),
"unionid": _user_info.get("user_id"),
"open_id": sender_open_id,
}
except Exception:
pass
# Resolve channel user via unified service (uses OrgMember + SSO patterns)
from app.services.channel_user_service import channel_user_service
platform_user = await channel_user_service.resolve_channel_user(
db=db,
agent=agent_obj,
channel_type="feishu",
external_user_id=sender_open_id,
extra_info=extra_info,
)
platform_user_id = platform_user.id
# Conv ID — prefer user_id for session continuity
if chat_type == "group" and chat_id:
conv_id = f"feishu_group_{chat_id}"
else:
conv_id = f"feishu_p2p_{sender_user_id_feishu or sender_open_id}"
# Find-or-create session
_is_group_file = (chat_type == "group")
# For group file sessions, use agent creator as placeholder user_id
_file_user_id = platform_user_id
if _is_group_file:
_ag_r = await db.execute(_select(AgentModel).where(AgentModel.id == agent_id))
_ag_obj = _ag_r.scalar_one_or_none()
_file_user_id = _ag_obj.creator_id if _ag_obj else platform_user_id
_sess = await find_or_create_channel_session(
db=db, agent_id=agent_id, user_id=_file_user_id,
external_conv_id=conv_id, source_channel="feishu",
first_message_title=f"[文件] {filename}",
is_group=_is_group_file,
group_name=f"Feishu Group {chat_id[:8]}" if _is_group_file else None,
)
session_conv_id = str(_sess.id)
# Store user message — include base64 marker for images so LLM can see them
if msg_type == "image":
import base64 as _b64_img
_b64_data = _b64_img.b64encode(file_bytes).decode("ascii")
_image_marker = f"[image_data:data:image/jpeg;base64,{_b64_data}]"
user_msg_content = f"[用户发送了图片]\n{_image_marker}"
else:
user_msg_content = f"[file:{filename}]"
db.add(ChatMessage(agent_id=agent_id, user_id=platform_user_id, role="user",
content=user_msg_content if msg_type != "image" else f"[file:{filename}]",
conversation_id=session_conv_id))
_sess.last_message_at = _dt.now(_tz.utc)
# Load conversation history for LLM context
from app.models.agent import DEFAULT_CONTEXT_WINDOW_SIZE
ctx_size = (agent_obj.context_window_size or DEFAULT_CONTEXT_WINDOW_SIZE) if agent_obj else DEFAULT_CONTEXT_WINDOW_SIZE
_hist_r = await db.execute(
_select(ChatMessage)
.where(ChatMessage.agent_id == agent_id, ChatMessage.conversation_id == session_conv_id)
.order_by(ChatMessage.created_at.desc())
.limit(ctx_size)
)
_history = [{"role": m.role, "content": m.content} for m in reversed(_hist_r.scalars().all())]
await db.commit()
# For images: call LLM so vision models can actually see the image
if msg_type == "image":
import json as _json_card_img
# Send initial loading card
_reply_to = chat_id if chat_type == "group" else sender_open_id
_rid_type = "chat_id" if chat_type == "group" else "open_id"
_agent_name = agent_obj.name if agent_obj else "AI"
_init_card = {
"config": {"update_multi": True},
"header": {"template": "blue", "title": {"content": "识别图片中...", "tag": "plain_text"}},
"elements": [{"tag": "markdown", "content": "..."}]
}
_patch_msg_id = None
try:
_init_resp = await feishu_service.send_message(
config.app_id, config.app_secret, _reply_to, "interactive",
_json_card_img.dumps(_init_card), receive_id_type=_rid_type, stage="image_stream_init_card"
)
_patch_msg_id = _init_resp.get("data", {}).get("message_id")
except Exception as _e_init:
logger.error(f"[Feishu] Failed to send init card for image: {_e_init}")
_img_stream_buf = []
_img_last_flush = time.time()
_img_flush_interval = 1.0
_img_patch_queue = _SerialPatchQueue()
_img_heartbeat_task: asyncio.Task | None = None
_img_llm_done = False
_img_last_flushed_hash: int = 0 # Content hash to skip no-op heartbeat patches
async def _queue_image_patch(_card: dict, _stage: str):
"""Enqueue a serialized PATCH request for the image streaming card."""
if not _patch_msg_id:
return
_payload = _json_card_img.dumps(_card)
async def _job():
try:
await feishu_service.patch_message(
config.app_id,
config.app_secret,
_patch_msg_id,
_payload,
stage=_stage,
)
except Exception as _e_patch:
logger.warning(f"[Feishu] Image patch failed (stage={_stage}, message_id={_patch_msg_id}): {_e_patch}")
_img_patch_queue.enqueue(_job)
async def _flush_image_stream(reason: str, force: bool = False):
"""Build and enqueue an image streaming card update.
Reuses _build_card so the image path supports the same thinking
and tool-status sections as the text streaming path.
Skips the patch on heartbeat ticks when content has not changed.
"""
nonlocal _img_last_flush, _img_last_flushed_hash
now = time.time()
if not force and now - _img_last_flush < _img_flush_interval:
return
# Reuse the shared card builder (no tool_status for image path yet,
# but the builder is ready to accept them in the future).
_card = _build_card(
"".join(_img_stream_buf),
streaming=True,
agent_name=_agent_name,
)
# Skip no-op heartbeat patches when content hasn't changed.
current_hash = hash("".join(_img_stream_buf))
if reason == "heartbeat" and current_hash == _img_last_flushed_hash:
return
_img_last_flushed_hash = current_hash
await _queue_image_patch(_card, _stage=f"image_stream_{reason}")
_img_last_flush = now
async def _img_on_chunk(text):
_img_stream_buf.append(text)
if _patch_msg_id:
await _flush_image_stream("chunk")
async def _img_heartbeat():
while not _img_llm_done:
await asyncio.sleep(_img_flush_interval)
if _patch_msg_id:
await _flush_image_stream("heartbeat")
if _patch_msg_id:
_img_heartbeat_task = asyncio.create_task(_img_heartbeat())
# Call LLM with image marker — vision models will parse it
async with _async_session() as _db_img:
try:
reply_text = await _call_agent_llm(
_db_img, agent_id, user_msg_content, history=_history,
user_id=platform_user_id, on_chunk=_img_on_chunk,
)
finally:
_img_llm_done = True
if _img_heartbeat_task:
_img_heartbeat_task.cancel()
try:
await _img_heartbeat_task
except Exception:
pass
logger.info(f"[Feishu] Image LLM reply: {reply_text[:100]}")
# Send final card or fallback text
if _patch_msg_id:
try:
await _img_patch_queue.drain()
except Exception as _e_drain:
logger.warning(f"[Feishu] Image patch queue drain failed: {_e_drain}")
# Build final card via shared builder (consistent with text streaming path).
_final_card = _build_card(
reply_text or "...",
streaming=False,
agent_name=_agent_name,
)
await feishu_service.patch_message(
config.app_id, config.app_secret, _patch_msg_id, _json_card_img.dumps(_final_card), stage="image_stream_final"
)
else:
try:
await feishu_service.send_message(
config.app_id, config.app_secret, _reply_to, "text",
json.dumps({"text": reply_text}), receive_id_type=_rid_type, stage="image_stream_fallback_text",
)
except Exception as _e_fb:
logger.error(f"[Feishu] Failed to send image reply: {_e_fb}")
# Save assistant reply in DB
async with _async_session() as _db_save:
_db_save.add(ChatMessage(agent_id=agent_id, user_id=platform_user_id, role="assistant",
content=reply_text, conversation_id=session_conv_id))
await _db_save.commit()
# Log activity
from app.services.activity_logger import log_activity
await log_activity(agent_id, "chat_reply", f"回复了飞书图片消息: {reply_text[:80]}", detail={"channel": "feishu", "type": "image"})
return
# For non-image files: send simple ack as before
await asyncio.sleep(random.uniform(1.0, 2.0))
ack = random.choice(_FILE_ACK_MESSAGES)
try:
if chat_type == "group" and chat_id:
await feishu_service.send_message(
config.app_id, config.app_secret, chat_id, "text",
json.dumps({"text": ack}), receive_id_type="chat_id",
)
else:
await feishu_service.send_message(
config.app_id, config.app_secret, sender_open_id, "text",
json.dumps({"text": ack}),
)
except Exception as e:
logger.error(f"[Feishu] Failed to send ack: {e}")
# Store ack in DB
async with _async_session() as db2:
db2.add(ChatMessage(agent_id=agent_id, user_id=platform_user_id, role="assistant",
content=ack, conversation_id=session_conv_id))
await db2.commit()
async def _download_post_images(agent_id, config, message_id, image_keys):
"""Download images embedded in a Feishu post message to the agent's workspace."""
from pathlib import Path
from app.config import get_settings
settings = get_settings()
upload_dir = Path(settings.AGENT_DATA_DIR) / str(agent_id) / "workspace" / "uploads"
upload_dir.mkdir(parents=True, exist_ok=True)
for ik in image_keys:
try:
file_bytes = await feishu_service.download_message_resource(
config.app_id, config.app_secret, message_id, ik, "image"
)
save_path = upload_dir / f"image_{ik[-8:]}.jpg"
save_path.write_bytes(file_bytes)
logger.info(f"[Feishu] Saved post image to {save_path} ({len(file_bytes)} bytes)")
except Exception as e:
logger.error(f"[Feishu] Failed to download post image {ik}: {e}")
async def _call_agent_llm(
db: AsyncSession,
agent_id: uuid.UUID,
user_text: str,
history: list[dict] | None = None,
user_id=None,
on_chunk=None,
on_thinking=None,
on_tool_call=None,
) -> str:
"""Call the agent's configured LLM model with conversation history.
Reuses the same call_llm function as the WebSocket chat endpoint so that
all providers (OpenRouter, Qwen, etc.) work identically on both channels.
"""
from app.models.agent import Agent
from app.models.llm import LLMModel
from app.api.websocket import call_llm
# Load agent and model
agent_result = await db.execute(select(Agent).where(Agent.id == agent_id))
agent = agent_result.scalar_one_or_none()
if not agent:
return "⚠️ 数字员工未找到"
if is_agent_expired(agent):
return "This Agent has expired and is off duty. Please contact your admin to extend its service."
# Load primary model (skip if disabled by admin)
model = None
if agent.primary_model_id:
model_result = await db.execute(select(LLMModel).where(LLMModel.id == agent.primary_model_id))
model = model_result.scalar_one_or_none()
if model and not model.enabled:
logger.info(f"[Channel] Primary model {model.model} is disabled, skipping")
model = None
# Load fallback model (skip if disabled by admin)
fallback_model = None
if agent.fallback_model_id:
fb_result = await db.execute(select(LLMModel).where(LLMModel.id == agent.fallback_model_id))
fallback_model = fb_result.scalar_one_or_none()
if fallback_model and not fallback_model.enabled:
logger.info(f"[Channel] Fallback model {fallback_model.model} is disabled, skipping")
fallback_model = None
# Config-level fallback: primary missing -> use fallback
if not model and fallback_model:
model = fallback_model
fallback_model = None
logger.warning(f"[Channel] Primary model unavailable, using fallback: {model.model}")
if not model:
return f"⚠️ {agent.name} 未配置 LLM 模型，请在管理后台设置。"
# Build conversation messages (without system prompt — call_llm adds it)
messages: list[dict] = []
from app.models.agent import DEFAULT_CONTEXT_WINDOW_SIZE
ctx_size = agent.context_window_size or DEFAULT_CONTEXT_WINDOW_SIZE
if history:
messages.extend(history[-ctx_size:])
messages.append({"role": "user", "content": user_text})
# Use actual user_id so the system prompt knows who it's chatting with
effective_user_id = user_id or agent_id
# Determine effective timeout: prefer model-level setting, else use module default.
_timeout = _get_llm_timeout(model)
try:
reply = await asyncio.wait_for(
call_llm(
model,
messages,
agent.name,
agent.role_description or "",
agent_id=agent_id,
user_id=effective_user_id,
supports_vision=getattr(model, 'supports_vision', False),
on_chunk=on_chunk,
on_thinking=on_thinking,
on_tool_call=on_tool_call,
),
timeout=_timeout,
)
return reply
except asyncio.TimeoutError:
logger.error(
f"[LLM] Call timed out after {_timeout}s "
f"(agent_id={agent_id}, model={getattr(model, 'model', 'unknown')})"
)
if fallback_model:
# Use the fallback model's own timeout budget.
_fb_timeout = _get_llm_timeout(fallback_model)
logger.info(f"[LLM] Retrying timed-out request with fallback model: {fallback_model.model} (timeout={_fb_timeout}s)")
try:
reply = await asyncio.wait_for(
call_llm(
fallback_model,
messages,
agent.name,
agent.role_description or "",
agent_id=agent_id,
user_id=effective_user_id,
supports_vision=getattr(fallback_model, 'supports_vision', False),
on_chunk=on_chunk,
on_thinking=on_thinking,
on_tool_call=on_tool_call,
),
timeout=_fb_timeout,
)
return reply
except asyncio.TimeoutError:
logger.error(
f"[LLM] Fallback call also timed out after {_fb_timeout}s "
f"(agent_id={agent_id}, model={getattr(fallback_model, 'model', 'unknown')})"
)
return f"⚠️ Model response timed out (>{int(_fb_timeout)}s). Please retry or shorten your request."
except Exception as e2:
import traceback
traceback.print_exc()
return f"⚠️ Model error: Primary Timeout | Fallback: {str(e2)[:80]}"
return f"⚠️ Model response timed out (>{int(_timeout)}s). Please retry or shorten your request."
except Exception as e:
import traceback
traceback.print_exc()
error_msg = str(e) or repr(e)
logger.error(f"[LLM] Primary model error: {error_msg}")
# Runtime fallback: primary model failed -> retry with fallback model
if fallback_model:
logger.info(f"[LLM] Retrying with fallback model: {fallback_model.model}")
try:
_fb_timeout = _get_llm_timeout(fallback_model)
reply = await asyncio.wait_for(
call_llm(
fallback_model,
messages,
agent.name,
agent.role_description or "",
agent_id=agent_id,
user_id=effective_user_id,
supports_vision=getattr(fallback_model, 'supports_vision', False),
on_chunk=on_chunk,
on_thinking=on_thinking,
on_tool_call=on_tool_call,
),
timeout=_fb_timeout,
)
return reply
except asyncio.TimeoutError:
logger.error(
f"[LLM] Fallback call timed out after {_fb_timeout}s "
f"(agent_id={agent_id}, model={getattr(fallback_model, 'model', 'unknown')})"
)
return f"⚠️ Model error: Primary: {str(e)[:80]} | Fallback Timeout"
except Exception as e2:
traceback.print_exc()
return f"⚠️ Model error: Primary: {str(e)[:80]} | Fallback: {str(e2)[:80]}"
return f"⚠️ 调用模型出错: {error_msg[:150]}"