feat: Feishu bridge for human-operator group chat

.env.feishu.example

@@ -0,0 +1,28 @@
+# Feishu bridge — copy to .env and fill in values.
+# Start: AGENT_MAILER_SECRET_KEY=... uv run agent-mailer-server feishu-bridge
+
+# Feishu app credentials (open.feishu.cn)
+FEISHU_APP_ID=
+FEISHU_APP_SECRET=
+FEISHU_VERIFICATION_TOKEN=
+FEISHU_ENCRYPT_KEY=
+
+# Target group chat id (oc_...)
+FEISHU_CHAT_ID=
+
+# Broker connection
+AGENT_MAILER_BASE_URL=http://127.0.0.1:9800
+AGENT_MAILER_OPERATOR_USER=
+AGENT_MAILER_OPERATOR_PASSWORD=
+
+# Fixed PM agent address in this user's namespace
+FEISHU_PM_ADDRESS=pm@youruser.amp.linkyun.co
+
+# Bridge process
+FEISHU_POLL_INTERVAL=3
+FEISHU_BRIDGE_HOST=0.0.0.0
+FEISHU_BRIDGE_PORT=9810
+FEISHU_BRIDGE_STATE_PATH=.feishu-bridge/state.json
+
+# Feishu event subscription callback URL (must be public):
+#   https://<your-host>/feishu/webhook

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "python-multipart>=0.0.9",
     "httpx>=0.28",
     "click>=8.1",
+    "cryptography>=42",
 ]
 
 [project.optional-dependencies]

src/agent_mailer/cli.py

@@ -167,6 +167,14 @@ def main():
     md = subparsers.add_parser("migrate-db", help="Migrate local-mode DB to SaaS mode")
     md.add_argument("--password", required=True, help="Password for the admin user")
 
+    # feishu-bridge
+    fb = subparsers.add_parser(
+        "feishu-bridge",
+        help="Start Feishu bridge (human-operator second screen for a group chat)",
+    )
+    fb.add_argument("--host", default=None, help="Bind host (default: FEISHU_BRIDGE_HOST or 0.0.0.0)")
+    fb.add_argument("--port", type=int, default=None, help="Bind port (default: FEISHU_BRIDGE_PORT or 9810)")
+
     args = parser.parse_args()
     if not args.command:
         parser.print_help()
@@ -178,6 +186,20 @@ def main():
         asyncio.run(_generate_invite_code(args))
     elif args.command == "migrate-db":
         asyncio.run(_migrate_db(args))
+    elif args.command == "feishu-bridge":
+        _feishu_bridge(args)
+
+
+def _feishu_bridge(args):
+    import uvicorn
+
+    from agent_mailer.feishu.bridge import create_app
+    from agent_mailer.feishu.config import FeishuBridgeConfig
+
+    config = FeishuBridgeConfig.from_env()
+    host = args.host or config.bridge_host
+    port = args.port or config.bridge_port
+    uvicorn.run(create_app(config), host=host, port=port)
 
 
 if __name__ == "__main__":

src/agent_mailer/feishu/__init__.py

@@ -0,0 +1 @@
+"""Feishu bridge — human-operator second screen for a Feishu group chat."""

src/agent_mailer/feishu/bridge.py

@@ -0,0 +1,204 @@
+"""FastAPI app: Feishu webhook (inbound) + broker inbox polling (outbound)."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+from typing import Any
+
+from fastapi import FastAPI, HTTPException, Request
+
+from agent_mailer.feishu.broker_client import BrokerClient
+from agent_mailer.feishu.config import FeishuBridgeConfig
+from agent_mailer.feishu.feishu_client import FeishuClient
+from agent_mailer.feishu.state import BridgeState
+
+logger = logging.getLogger(__name__)
+
+
+def _normalize_headers(headers: Any) -> dict[str, str]:
+    return {k.lower(): v for k, v in headers.items()}
+
+
+async def poll_outbound_once(
+    broker: BrokerClient,
+    feishu: FeishuClient,
+    state: BridgeState,
+    config: FeishuBridgeConfig,
+) -> None:
+    messages = await broker.inbox_unread()
+    for msg in reversed(messages):
+        msg_id = msg["id"]
+        if state.is_broker_message_pushed(msg_id):
+            continue
+        text = FeishuClient.format_outbound_message(msg)
+        await feishu.send_text(config.chat_id, text)
+        await broker.mark_read(msg_id)
+        state.record_broker_message_pushed(msg_id)
+        state.set_active_thread(msg.get("thread_id"))
+
+
+async def outbound_poll_loop(app: FastAPI) -> None:
+    broker: BrokerClient = app.state.broker
+    feishu: FeishuClient = app.state.feishu
+    state: BridgeState = app.state.state
+    config: FeishuBridgeConfig = app.state.config
+
+    while True:
+        try:
+            await poll_outbound_once(broker, feishu, state, config)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("Outbound poll failed")
+        await asyncio.sleep(config.poll_interval)
+
+
+async def handle_inbound_message(
+    broker: BrokerClient,
+    feishu: FeishuClient,
+    state: BridgeState,
+    config: FeishuBridgeConfig,
+    event: dict[str, Any],
+) -> None:
+    message = event.get("message") or {}
+    message_id = message.get("message_id")
+    if not message_id:
+        return
+
+    if message.get("chat_id") != config.chat_id:
+        return
+
+    sender = event.get("sender") or {}
+    if sender.get("sender_type") == "app":
+        return
+
+    if message.get("message_type") != "text":
+        return
+
+    if not await feishu.is_bot_mentioned(message):
+        return
+
+    if state.is_feishu_message_seen(message_id):
+        return
+
+    text = FeishuClient.extract_text(message)
+    if not text:
+        return
+
+    state.record_feishu_message(message_id)
+
+    parent_id: str | None = None
+    action = "send"
+    subject = "Feishu"
+
+    thread_id = state.active_thread_id
+    if thread_id:
+        thread_msgs = await broker.get_thread(thread_id)
+        if thread_msgs:
+            parent_id = thread_msgs[-1]["id"]
+            action = "reply"
+            subject = ""
+
+    result = await broker.send_message(
+        to_agent=config.pm_address,
+        action=action,
+        subject=subject,
+        body=text,
+        parent_id=parent_id,
+    )
+    state.set_active_thread(result.get("thread_id"))
+
+
+def create_app(
+    config: FeishuBridgeConfig | None = None,
+    *,
+    enable_poll: bool = True,
+    broker: BrokerClient | None = None,
+    feishu: FeishuClient | None = None,
+    state: BridgeState | None = None,
+) -> FastAPI:
+    bridge_config = config or FeishuBridgeConfig.from_env()
+
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        app.state.config = bridge_config
+        app.state.broker = broker or BrokerClient(
+            bridge_config.broker_base_url,
+            bridge_config.operator_username,
+            bridge_config.operator_password,
+        )
+        app.state.feishu = feishu or FeishuClient(
+            bridge_config.app_id,
+            bridge_config.app_secret,
+            bridge_config.verification_token,
+            bridge_config.encrypt_key,
+        )
+        app.state.state = state or BridgeState(bridge_config.state_path)
+
+        if broker is None:
+            await app.state.broker.login()
+            await app.state.broker.ensure_human_operator()
+        if feishu is None:
+            await app.state.feishu.bot_open_id()
+
+        poll_task = None
+        if enable_poll:
+            poll_task = asyncio.create_task(outbound_poll_loop(app))
+        logger.info(
+            "Feishu bridge started (broker=%s, chat=%s, pm=%s)",
+            bridge_config.broker_base_url,
+            bridge_config.chat_id,
+            bridge_config.pm_address,
+        )
+        try:
+            yield
+        finally:
+            if poll_task is not None:
+                poll_task.cancel()
+                try:
+                    await poll_task
+                except asyncio.CancelledError:
+                    pass
+            await app.state.broker.aclose()
+            await app.state.feishu.aclose()
+
+    app = FastAPI(title="Agent Mailer Feishu Bridge", lifespan=lifespan)
+
+    @app.get("/health")
+    async def health():
+        return {"status": "ok"}
+
+    @app.post("/feishu/webhook")
+    async def feishu_webhook(request: Request):
+        body_bytes = await request.body()
+        body = body_bytes.decode("utf-8")
+        feishu: FeishuClient = request.app.state.feishu
+        broker: BrokerClient = request.app.state.broker
+        state: BridgeState = request.app.state.state
+        config: FeishuBridgeConfig = request.app.state.config
+
+        try:
+            payload = feishu.parse_event(_normalize_headers(request.headers), body)
+        except ValueError as exc:
+            raise HTTPException(status_code=403, detail=str(exc)) from exc
+
+        if payload.get("type") == "url_verification":
+            return {"challenge": payload["challenge"]}
+
+        header = payload.get("header") or {}
+        if header.get("event_type") != "im.message.receive_v1":
+            return {"ok": True}
+
+        try:
+            await handle_inbound_message(
+                broker, feishu, state, config, payload.get("event") or {}
+            )
+        except Exception:
+            logger.exception("Inbound Feishu message handling failed")
+            raise HTTPException(status_code=500, detail="Inbound handling failed") from None
+
+        return {"ok": True}
+
+    return app

src/agent_mailer/feishu/broker_client.py

@@ -0,0 +1,97 @@
+"""HTTP client for broker admin routes (session-authenticated)."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+
+class BrokerClient:
+    def __init__(self, base_url: str, username: str, password: str):
+        self.base_url = base_url.rstrip("/")
+        self.username = username
+        self.password = password
+        self._client = httpx.AsyncClient(base_url=self.base_url, timeout=30.0)
+        self._session_token: str | None = None
+        self._op_address: str | None = None
+
+    async def aclose(self) -> None:
+        await self._client.aclose()
+
+    def _auth_headers(self) -> dict[str, str]:
+        if not self._session_token:
+            return {}
+        return {"Authorization": f"Bearer {self._session_token}"}
+
+    async def login(self) -> None:
+        resp = await self._client.post(
+            "/users/login",
+            json={"username": self.username, "password": self.password},
+        )
+        resp.raise_for_status()
+        self._session_token = resp.json()["token"]
+        logger.info("Broker session established for user %s", self.username)
+
+    async def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
+        extra_headers = dict(kwargs.pop("headers", {}))
+        headers = {**extra_headers, **self._auth_headers()}
+        resp = await self._client.request(method, path, headers=headers, **kwargs)
+        if resp.status_code == 401:
+            logger.warning("Broker session expired; re-authenticating")
+            await self.login()
+            headers = {**extra_headers, **self._auth_headers()}
+            resp = await self._client.request(method, path, headers=headers, **kwargs)
+        resp.raise_for_status()
+        return resp
+
+    async def ensure_human_operator(self) -> dict[str, str]:
+        resp = await self._request("GET", "/admin/human-operator")
+        data = resp.json()
+        self._op_address = data["address"]
+        return data
+
+    @property
+    def op_address(self) -> str:
+        if not self._op_address:
+            raise RuntimeError("human-operator address not loaded; call ensure_human_operator()")
+        return self._op_address
+
+    async def inbox_unread(self, address: str | None = None) -> list[dict[str, Any]]:
+        addr = address or self.op_address
+        resp = await self._request("GET", f"/admin/messages/inbox/{addr}")
+        data = resp.json()
+        if isinstance(data, dict) and "messages" in data:
+            return data["messages"]
+        return data
+
+    async def mark_read(self, message_id: str) -> dict[str, Any]:
+        resp = await self._request("PATCH", f"/admin/messages/{message_id}/read")
+        return resp.json()
+
+    async def send_message(
+        self,
+        *,
+        to_agent: str,
+        action: str,
+        subject: str,
+        body: str,
+        parent_id: str | None = None,
+    ) -> dict[str, Any]:
+        payload: dict[str, Any] = {
+            "to_agent": to_agent,
+            "action": action,
+            "subject": subject,
+            "body": body,
+        }
+        if parent_id:
+            payload["parent_id"] = parent_id
+        resp = await self._request("POST", "/admin/messages/send", json=payload)
+        return resp.json()
+
+    async def get_thread(self, thread_id: str) -> list[dict[str, Any]]:
+        resp = await self._request("GET", f"/admin/messages/thread/{thread_id}")
+        return resp.json()