diff --git a/backend/package/yuxi/channel/extensions/farcaster/__init__.py b/backend/package/yuxi/channel/extensions/farcaster/__init__.py
new file mode 100644
index 00000000..026c46a6
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/__init__.py
@@ -0,0 +1,503 @@
+import logging
+
+import httpx
+
+from yuxi.channel.capabilities import ChannelCapabilities
+from yuxi.channel.context import ChannelContext
+from yuxi.channel.extensions.base import BaseChannelPlugin
+from yuxi.channel.extensions.farcaster.client import NeynarClient, NeynarError
+from yuxi.channel.extensions.farcaster.config import FarcasterConfigAdapter
+from yuxi.channel.extensions.farcaster.defaults import DEDUPE_TTL_S, DEDUPE_MAX_TRACKED
+from yuxi.channel.extensions.farcaster.dedupe import FarcasterDeduplicator
+from yuxi.channel.extensions.farcaster.direct_cast_client import DirectCastClient, DirectCastError
+from yuxi.channel.extensions.farcaster.gateway import FarcasterGateway, FarcasterGatewayError
+from yuxi.channel.extensions.farcaster.outbound import FarcasterOutboundAdapter
+from yuxi.channel.extensions.farcaster.webhook import (
+    register_gateway_state,
+    unregister_gateway_state,
+    _dispatch_webhook_event,
+)
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+from yuxi.channel.protocols import ClassifiedError, ErrorSeverity
+
+logger = logging.getLogger(__name__)
+
+
+class FarcasterPlugin(BaseChannelPlugin):
+    id = "farcaster"
+    name = "Farcaster"
+    order = 105
+    label = "Farcaster"
+    aliases = ["fc"]
+    resolve_reply_to_mode = "thread"
+
+    def __init__(self):
+        super().__init__()
+        self._config = FarcasterConfigAdapter()
+        self._gateway = FarcasterGateway()
+        self._outbound: FarcasterOutboundAdapter | None = None
+        self._dedupe = FarcasterDeduplicator()
+
+    @property
+    def capabilities(self) -> ChannelCapabilities:
+        return ChannelCapabilities(
+            chat_types=["direct"],
+            message_types=["text"],
+            interactions=["recast"],
+            reactions=True,
+            typing_indicator=False,
+            threads=True,
+            edit=False,
+            unsend=True,
+            reply=True,
+            media=True,
+            native_commands=False,
+            polls=False,
+            streaming=True,
+            streaming_mode="block",
+            block_streaming=True,
+            block_streaming_chunk_min_chars=240,
+            block_streaming_chunk_max_chars=300,
+            block_streaming_chunk_break_preference="paragraph",
+        )
+
+    # ── ConfigProtocol ──────────────────────────────────
+
+    def list_account_ids(self, config: dict) -> list[str]:
+        return self._config.list_account_ids(config)
+
+    async def resolve_account(self, account_id: str) -> dict:
+        return await self._config.resolve_account(account_id)
+
+    def is_configured(self, account: dict) -> bool:
+        return self._config.is_configured(account)
+
+    def is_enabled(self, account: dict) -> bool:
+        return self._config.is_enabled(account)
+
+    def disabled_reason(self, account: dict) -> str:
+        return self._config.disabled_reason(account)
+
+    def describe_account(self, account: dict) -> dict:
+        return self._config.describe_account(account)
+
+    async def resolve_allow_from(self, config: dict, account_id: str) -> list[str] | None:
+        account = await self._config.resolve_account(account_id)
+        return account.get("allow_from", [])
+
+    def config_schema(self) -> dict:
+        return self._config.config_schema()
+
+    # ── GatewayProtocol ─────────────────────────────────
+
+    async def start(self, ctx: ChannelContext) -> object:
+        account_id = getattr(ctx, "account_id", "default")
+        self._config.update_config(ctx.config)
+        account = await self._config.resolve_account(account_id)
+
+        if not account.get("configured"):
+            raise FarcasterGatewayError("Farcaster account not configured: missing api_key")
+
+        self._outbound = FarcasterOutboundAdapter(self._client_getter, account_id=account_id)
+
+        handle = await self._gateway.start(ctx)
+
+        register_gateway_state(account_id, ctx, self._gateway, self._dedupe)
+
+        logger.info(
+            "Farcaster plugin started: account=%s fid=%s fname=%s",
+            account_id,
+            handle.get("bot_fid"),
+            handle.get("bot_fname"),
+        )
+        return handle
+
+    async def stop(self, ctx: ChannelContext) -> None:
+        account_id = getattr(ctx, "account_id", "default")
+        unregister_gateway_state(account_id)
+        await self._gateway.stop(ctx)
+        self._outbound = None
+
+    # ── OutboundProtocol ────────────────────────────────
+
+    @property
+    def delivery_mode(self):
+        if self._outbound is None:
+            from yuxi.channel.protocols import OutboundDeliveryMode
+
+            return OutboundDeliveryMode.DIRECT
+        return self._outbound.delivery_mode
+
+    @property
+    def text_chunk_limit(self) -> int | None:
+        if self._outbound is None:
+            return 320
+        return self._outbound.text_chunk_limit
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        if self._outbound is None:
+            logger.error("Farcaster send_text: outbound not initialized")
+            return
+        resolved = await self._config.resolve_account(account_id or "default")
+        channel_id = resolved.get("channel_id", "")
+        await self._outbound.send_text(
+            target_id,
+            content,
+            reply_to_id=reply_to_id,
+            thread_id=thread_id,
+            account_id=account_id,
+            channel_id=channel_id or None,
+        )
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        if self._outbound is None:
+            return
+        await self._outbound.send_media(
+            target_id,
+            media_url,
+            media_type,
+            reply_to_id=reply_to_id,
+            thread_id=thread_id,
+            account_id=account_id,
+        )
+
+    async def send_recast(
+        self,
+        target_hash: str,
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        if self._outbound is None:
+            logger.error("Farcaster send_recast: outbound not initialized")
+            return
+        await self._outbound.send_reaction(target_hash, reaction_type="recast")
+
+    async def delete_cast(
+        self,
+        cast_hash: str,
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            logger.error("Farcaster delete_cast: no active client")
+            return
+        try:
+            await client.delete_cast(cast_hash)
+            logger.info("Farcaster cast deleted: hash=%s", cast_hash)
+        except NeynarError as e:
+            logger.error("Farcaster delete_cast failed: %s", e)
+
+    async def send_direct_cast(
+        self,
+        recipient_fid: int,
+        content: str,
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        account = await self._config.resolve_account(account_id or "default")
+        api_key = account.get("api_key", "")
+        signer_uuid = account.get("signer_uuid", "")
+        if not api_key or not signer_uuid:
+            logger.error("Farcaster send_direct_cast: missing api_key or signer_uuid")
+            return
+        dc_client = DirectCastClient(api_key=api_key, signer_uuid=signer_uuid)
+        try:
+            await dc_client.send_direct_cast(recipient_fid, content)
+            logger.info("Farcaster direct cast sent to fid=%s", recipient_fid)
+        except DirectCastError as e:
+            logger.error("Farcaster send_direct_cast failed: %s", e)
+        finally:
+            await dc_client.close()
+
+    async def follow_user(
+        self,
+        target_fid: int,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return await client.follow_user(target_fid)
+
+    async def unfollow_user(
+        self,
+        target_fid: int,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return await client.unfollow_user(target_fid)
+
+    async def search_users(
+        self,
+        query: str,
+        *,
+        limit: int = 10,
+        account_id: str | None = None,
+    ) -> list[dict]:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return await client.search_users(query, limit=limit)
+
+    async def block_user(
+        self,
+        target_fid: int,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return await client.block_user(target_fid)
+
+    async def unblock_user(
+        self,
+        target_fid: int,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return await client.unblock_user(target_fid)
+
+    async def mute_user(
+        self,
+        target_fid: int,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return await client.mute_user(target_fid)
+
+    async def unmute_user(
+        self,
+        target_fid: int,
+        *,
+        account_id: str | None = None,
+    ) -> dict:
+        client = self._client_getter(account_id or "default")
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return await client.unmute_user(target_fid)
+
+    # ── StatusProtocol ──────────────────────────────────
+
+    async def probe(self, account: dict | None = None) -> bool:
+        api_key = ""
+        if account:
+            api_key = account.get("api_key", "")
+        else:
+            resolved = await self._config.resolve_account("default")
+            api_key = resolved.get("api_key", "")
+        if not api_key:
+            return False
+
+        client = NeynarClient(api_key=api_key)
+        try:
+            await client.get_me()
+            return True
+        except NeynarError:
+            return False
+        finally:
+            await client.close()
+
+    def build_summary(self, snapshot: object) -> dict:
+        return {"channel": "farcaster", "accounts": self._gateway.account_count}
+
+    # ── SecurityProtocol ────────────────────────────────
+
+    async def check_allowlist(self, peer_id: str, channel_type: str, account_id: str = "default") -> bool:
+        account = await self._config.resolve_account(account_id)
+        allow_from = account.get("allow_from", [])
+        if not allow_from:
+            return False
+        fid = peer_id.replace("farcaster:", "").strip()
+        return fid in allow_from or str(fid) in allow_from
+
+    def resolve_dm_policy(self) -> dict:
+        return {"mode": "pairing", "allow_from": []}
+
+    # ── AgentPromptProtocol ─────────────────────────────
+
+    @property
+    def channel_format_instructions(self) -> str | None:
+        return (
+            "You are connected to Farcaster, a decentralized social protocol. "
+            "Messages are public Casts visible to everyone on the network. "
+            "Each Cast is limited to 320 UTF-8 bytes. Longer responses will be automatically "
+            "split into multiple Cast replies in the same thread. "
+            "No markdown formatting is supported — use plain text only. "
+            "URLs placed in Casts are automatically rendered as embeds (images, links) by clients. "
+            "Use @username to mention other users. "
+            "Farcaster community appreciates concise, Web3-native communication style. "
+            "Keep answers brief, direct, and informative."
+        )
+
+    # ── DedupeProtocol ──────────────────────────────────
+
+    def is_duplicate(self, key: str) -> bool:
+        return self._dedupe.is_duplicate(key)
+
+    def mark_seen(self, key: str) -> None:
+        self._dedupe.is_duplicate(key)
+
+    def reset(self) -> None:
+        self._dedupe.reset()
+
+    @property
+    def ttl_seconds(self) -> int:
+        return DEDUPE_TTL_S
+
+    @property
+    def max_entries(self) -> int:
+        return DEDUPE_MAX_TRACKED
+
+    # ── InboundHandlerProtocol ──────────────────────────
+
+    def resolve_event_type(self, raw_event: dict) -> str:
+        return raw_event.get("type", "unknown")
+
+    def parse_to_unified(
+        self,
+        raw_event: dict,
+        account_id: str,
+    ) -> object | None:
+        from yuxi.channel.extensions.farcaster.monitor import FarcasterMonitor
+
+        event_type = raw_event.get("type", "")
+        event_data = raw_event.get("data", {})
+
+        handle = self._gateway.get_handle(account_id)
+        bot_fid = handle.get("bot_fid", 0) if handle else 0
+        bot_fname = handle.get("bot_fname", "") if handle else ""
+
+        monitor = FarcasterMonitor(bot_fid=bot_fid, bot_fname=bot_fname)
+
+        if event_type == "cast.created":
+            return monitor.convert_cast_to_unified(event_data, account_id=account_id)
+        elif event_type == "reaction.created":
+            return monitor.convert_reaction_to_unified(event_data, account_id=account_id)
+        return None
+
+    async def handle_raw_event(
+        self,
+        event: dict,
+        account: dict,
+    ) -> object | None:
+        return self.parse_to_unified(event, account.get("account_id", "default"))
+
+    # ── MentionsProtocol ────────────────────────────────
+
+    def extract_mentions(self, raw_message: dict) -> list[str]:
+        from yuxi.channel.extensions.farcaster.monitor import FarcasterMonitor
+
+        text = raw_message.get("text", "")
+        if not text:
+            return []
+        monitor = FarcasterMonitor()
+        return monitor.extract_mentions(text)
+
+    # ── FormatProtocol ──────────────────────────────────
+
+    def markdown_to_native(self, md_text: str) -> dict | str:
+        return _strip_markdown_formatting(md_text)
+
+    def native_to_markdown(self, native_content: dict | str) -> str:
+        if isinstance(native_content, dict):
+            return native_content.get("text", "")
+        return str(native_content)
+
+    # ── WebhookProtocol ─────────────────────────────────
+
+    def verify_signature(self, body: bytes, signature: str, secret: str) -> bool:
+        import hashlib
+        import hmac
+
+        expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+        return hmac.compare_digest(expected, signature)
+
+    async def handle_webhook(
+        self,
+        request: object,
+        account_id: str | None = None,
+    ) -> object:
+        return await _dispatch_webhook_event(request, account_id or "default")
+
+    def resolve_webhook_auth_bypass_paths(self) -> list[str]:
+        return ["/farcaster/webhook"]
+
+    # ── ErrorHandlingProtocol ───────────────────────────
+
+    def classify_error(self, error: BaseException) -> ClassifiedError:
+        severity = self._classify_severity(error)
+        return ClassifiedError(severity=severity, original_error=error, error_message=str(error))
+
+    def is_retryable(self, error: BaseException) -> bool:
+        return self._classify_severity(error) == ErrorSeverity.RETRYABLE
+
+    def _classify_severity(self, error: BaseException) -> ErrorSeverity:
+        if isinstance(error, NeynarError):
+            return ErrorSeverity.RETRYABLE
+        if isinstance(error, FarcasterGatewayError):
+            return ErrorSeverity.FATAL
+        if isinstance(error, httpx.HTTPStatusError):
+            status = error.response.status_code
+            if status == 429:
+                return ErrorSeverity.RATE_LIMITED
+            if status in (401, 403):
+                return ErrorSeverity.FORBIDDEN
+            if status >= 500:
+                return ErrorSeverity.RETRYABLE
+            return ErrorSeverity.FATAL
+        if isinstance(error, (httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError)):
+            return ErrorSeverity.NETWORK
+        classified = super().classify_error(error)
+        return classified.severity if hasattr(classified, "severity") else ErrorSeverity.FATAL
+
+    # ── Helpers ─────────────────────────────────────────
+
+    def _client_getter(self, account_id: str = "default"):
+        handle = self._gateway.get_handle(account_id)
+        return handle.get("client") if handle else None
+
+
+def _strip_markdown_formatting(text: str) -> str:
+    import re
+
+    text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
+    text = re.sub(r"\*(.+?)\*", r"\1", text)
+    text = re.sub(r"~~(.+?)~~", r"\1", text)
+    text = re.sub(r"`([^`]+)`", r"\1", text)
+    text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE)
+    text = re.sub(r"^[-*+]\s+", "", text, flags=re.MULTILINE)
+    text = re.sub(r"^>\s?", "", text, flags=re.MULTILINE)
+    text = re.sub(r"!?\[([^\]]*)\]\([^)]+\)", r"\1", text)
+    return text.strip()
+
+
+farcaster_plugin = FarcasterPlugin()
+ChannelPluginRegistry.register(farcaster_plugin)
diff --git a/backend/package/yuxi/channel/extensions/farcaster/client.py b/backend/package/yuxi/channel/extensions/farcaster/client.py
new file mode 100644
index 00000000..69a1aa49
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/client.py
@@ -0,0 +1,341 @@
+import asyncio
+import hashlib
+import hmac
+import secrets
+import uuid
+from functools import wraps
+
+import httpx
+
+from yuxi.channel.extensions.farcaster.defaults import (
+    CLIENT_TIMEOUT_S,
+    RETRY_MAX_ATTEMPTS,
+    RETRY_BASE_DELAY_S,
+    RETRY_MAX_DELAY_S,
+)
+
+NEYNAR_BASE_URL = "https://api.neynar.com/v2/farcaster"
+
+logger = __import__("logging").getLogger(__name__)
+
+
+class NeynarError(Exception):
+    pass
+
+
+def _retry_on_network_error(max_attempts: int = RETRY_MAX_ATTEMPTS):
+    def decorator(func):
+        @wraps(func)
+        async def wrapper(self, *args, **kwargs):
+            last_exc = None
+            for attempt in range(1, max_attempts + 1):
+                try:
+                    return await func(self, *args, **kwargs)
+                except httpx.HTTPStatusError as e:
+                    if e.response.status_code < 500:
+                        raise
+                    last_exc = e
+                except (httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError) as e:
+                    last_exc = e
+
+                if attempt < max_attempts:
+                    delay = min(RETRY_BASE_DELAY_S * (2 ** (attempt - 1)), RETRY_MAX_DELAY_S)
+                    logger.warning(
+                        "NeynarClient %s attempt %d/%d failed, retrying in %.1fs: %s",
+                        func.__name__,
+                        attempt,
+                        max_attempts,
+                        delay,
+                        last_exc,
+                    )
+                    await asyncio.sleep(delay)
+
+            raise last_exc or NeynarError("Retry exhausted")
+
+        return wrapper
+
+    return decorator
+
+
+class NeynarClient:
+    def __init__(
+        self,
+        api_key: str,
+        signer_uuid: str | None = None,
+        webhook_secret: str | None = None,
+        timeout: float = CLIENT_TIMEOUT_S,
+    ):
+        self._api_key = api_key
+        self._signer_uuid = signer_uuid
+        self._webhook_secret = webhook_secret or secrets.token_hex(32)
+        self._client: httpx.AsyncClient | None = None
+        self._timeout = timeout
+
+    async def _ensure_client(self) -> httpx.AsyncClient:
+        if self._client is None:
+            self._client = httpx.AsyncClient(
+                base_url=NEYNAR_BASE_URL,
+                headers={
+                    "accept": "application/json",
+                    "x-api-key": self._api_key,
+                },
+                timeout=httpx.Timeout(self._timeout),
+            )
+        return self._client
+
+    async def get_me(self) -> dict:
+        client = await self._ensure_client()
+        resp = await client.get("/user/me")
+        resp.raise_for_status()
+        return resp.json()["result"]["user"]
+
+    async def get_user_by_fid(self, fid: int) -> dict:
+        client = await self._ensure_client()
+        resp = await client.get("/user/bulk", params={"fids": str(fid)})
+        resp.raise_for_status()
+        users = resp.json().get("users", [])
+        if not users:
+            raise NeynarError(f"User not found: fid={fid}")
+        return users[0]
+
+    async def get_user_by_fname(self, fname: str) -> dict:
+        client = await self._ensure_client()
+        resp = await client.get("/user/bulk", params={"fnames": fname})
+        resp.raise_for_status()
+        users = resp.json().get("users", [])
+        if not users:
+            raise NeynarError(f"User not found: fname={fname}")
+        return users[0]
+
+    async def get_cast(self, cast_hash: str) -> dict:
+        client = await self._ensure_client()
+        resp = await client.get("/cast", params={"identifier": cast_hash, "type": "hash"})
+        resp.raise_for_status()
+        return resp.json()
+
+    async def get_notifications(self, fid: int, cursor: str | None = None, limit: int = 25) -> dict:
+        client = await self._ensure_client()
+        params = {"fid": fid, "limit": limit}
+        if cursor:
+            params["cursor"] = cursor
+        resp = await client.get("/notifications/channel", params=params)
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def post_cast(
+        self,
+        text: str,
+        *,
+        parent_hash: str | None = None,
+        embeds: list[dict] | None = None,
+        channel_id: str | None = None,
+    ) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+
+        body: dict = {"signer_uuid": self._signer_uuid, "text": text, "idem": str(uuid.uuid4())}
+        if parent_hash:
+            body["parent"] = parent_hash
+        if embeds:
+            body["embeds"] = embeds
+        if channel_id:
+            body["channel_id"] = channel_id
+
+        client = await self._ensure_client()
+        resp = await client.post("/cast", json=body)
+        resp.raise_for_status()
+        data = resp.json()
+        if not data.get("success"):
+            raise NeynarError(f"Cast publish failed: {data}")
+        return data
+
+    @_retry_on_network_error()
+    async def delete_cast(self, cast_hash: str) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+
+        client = await self._ensure_client()
+        resp = await client.delete(
+            "/cast",
+            json={"signer_uuid": self._signer_uuid, "target_hash": cast_hash},
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def post_reaction(self, target_hash: str, reaction_type: str = "like") -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+
+        body = {
+            "signer_uuid": self._signer_uuid,
+            "type": reaction_type,
+            "target": target_hash,
+            "idem": str(uuid.uuid4()),
+        }
+        client = await self._ensure_client()
+        resp = await client.post("/reaction", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def delete_reaction(self, target_hash: str, reaction_type: str = "like") -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+
+        body = {
+            "signer_uuid": self._signer_uuid,
+            "type": reaction_type,
+            "target": target_hash,
+        }
+        client = await self._ensure_client()
+        resp = await client.delete("/reaction", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    async def register_webhook(self, url: str, filters: dict) -> dict:
+        body = {
+            "name": "ForcePilot Farcaster Bot",
+            "url": url,
+            "webhook_secret": self._webhook_secret,
+            "subscription": filters,
+        }
+        client = await self._ensure_client()
+        resp = await client.post("/webhook", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    async def get_webhooks(self) -> list[dict]:
+        client = await self._ensure_client()
+        resp = await client.get("/webhook")
+        resp.raise_for_status()
+        return resp.json().get("webhooks", [])
+
+    async def delete_webhook(self, webhook_id: str) -> None:
+        client = await self._ensure_client()
+        resp = await client.delete("/webhook", params={"webhook_id": webhook_id})
+        resp.raise_for_status()
+
+    async def validate_frame(self, message_bytes: str) -> dict:
+        body = {"message_bytes_in_hex": message_bytes}
+        client = await self._ensure_client()
+        resp = await client.post("/frame/validate", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def follow_user(self, target_fid: int) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+        client = await self._ensure_client()
+        body = {
+            "signer_uuid": self._signer_uuid,
+            "target_fid": target_fid,
+            "idem": str(uuid.uuid4()),
+        }
+        resp = await client.post("/user/follow", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def unfollow_user(self, target_fid: int) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+        client = await self._ensure_client()
+        resp = await client.delete(
+            "/user/follow",
+            json={
+                "signer_uuid": self._signer_uuid,
+                "target_fid": target_fid,
+            },
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    async def get_storage_usage(self, fid: int) -> dict:
+        client = await self._ensure_client()
+        resp = await client.get("/storage/usage", params={"fid": fid})
+        resp.raise_for_status()
+        return resp.json()
+
+    async def search_users(self, query: str, limit: int = 10) -> list[dict]:
+        client = await self._ensure_client()
+        params: dict = {"q": query}
+        if limit:
+            params["limit"] = limit
+        resp = await client.get("/user/search", params=params)
+        resp.raise_for_status()
+        result = resp.json()
+        return result.get("result", {}).get("users", [])
+
+    @_retry_on_network_error()
+    async def block_user(self, target_fid: int) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+        client = await self._ensure_client()
+        body = {
+            "signer_uuid": self._signer_uuid,
+            "blocked_fid": target_fid,
+            "idem": str(uuid.uuid4()),
+        }
+        resp = await client.post("/block", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def unblock_user(self, target_fid: int) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+        client = await self._ensure_client()
+        body = {
+            "signer_uuid": self._signer_uuid,
+            "blocked_fid": target_fid,
+        }
+        resp = await client.delete("/block", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def mute_user(self, target_fid: int) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+        client = await self._ensure_client()
+        body = {
+            "signer_uuid": self._signer_uuid,
+            "muted_fid": target_fid,
+            "idem": str(uuid.uuid4()),
+        }
+        resp = await client.post("/mute", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    @_retry_on_network_error()
+    async def unmute_user(self, target_fid: int) -> dict:
+        if not self._signer_uuid:
+            raise NeynarError("signer_uuid required for write operations")
+        client = await self._ensure_client()
+        body = {
+            "signer_uuid": self._signer_uuid,
+            "muted_fid": target_fid,
+        }
+        resp = await client.delete("/mute", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    def verify_webhook_signature(self, body: bytes, signature: str) -> bool:
+        expected = hmac.new(
+            self._webhook_secret.encode(),
+            body,
+            hashlib.sha256,
+        ).hexdigest()
+        return hmac.compare_digest(expected, signature)
+
+    @property
+    def webhook_secret(self) -> str:
+        return self._webhook_secret
+
+    async def close(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
diff --git a/backend/package/yuxi/channel/extensions/farcaster/config.py b/backend/package/yuxi/channel/extensions/farcaster/config.py
new file mode 100644
index 00000000..a9709a9b
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/config.py
@@ -0,0 +1,172 @@
+import logging
+import os
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+ENV_NEYNAR_API_KEY = "NEYNAR_API_KEY"
+ENV_NEYNAR_SIGNER_UUID = "NEYNAR_SIGNER_UUID"
+
+
+class FarcasterConfigAdapter:
+    def __init__(self):
+        self._config: dict = {}
+
+    def update_config(self, config: dict) -> None:
+        self._config = config
+
+    def list_account_ids(self, config: dict) -> list[str]:
+        self._config = config
+        fc_cfg = self._resolve_fc_config(config)
+        accounts = fc_cfg.get("accounts", {}) if isinstance(fc_cfg, dict) else {}
+        if accounts:
+            return list(accounts.keys())
+        return ["default"]
+
+    async def resolve_account(self, account_id: str) -> dict:
+        fc_cfg = self._get_farcaster_config()
+        accounts = fc_cfg.get("accounts", {}) if isinstance(fc_cfg, dict) else {}
+        raw = accounts.get(account_id, {}) if isinstance(accounts, dict) else {}
+        base = {k: v for k, v in fc_cfg.items() if k != "accounts"}
+
+        api_key = (
+            raw.get("api_key")
+            or base.get("api_key", "")
+            or _read_file(raw.get("api_key_file", "") or base.get("api_key_file", ""))
+            or os.environ.get(ENV_NEYNAR_API_KEY, "")
+        )
+        signer_uuid = (
+            raw.get("signer_uuid") or base.get("signer_uuid", "") or os.environ.get(ENV_NEYNAR_SIGNER_UUID, "")
+        )
+
+        return {
+            "account_id": account_id,
+            "name": raw.get("name", account_id),
+            "enabled": raw.get("enabled", base.get("enabled", True)),
+            "configured": bool(api_key),
+            "api_key": api_key,
+            "signer_uuid": signer_uuid,
+            "fid": raw.get("fid", base.get("fid", 0)),
+            "fname": raw.get("fname", base.get("fname", "")),
+            "dm_policy": raw.get("dm_policy", base.get("dm_policy", "pairing")),
+            "allow_from": raw.get("allow_from", base.get("allow_from", [])),
+            "webhook_base_url": raw.get("webhook_base_url", base.get("webhook_base_url", "")),
+            "channel_id": raw.get("channel_id", base.get("channel_id", "")),
+        }
+
+    def is_configured(self, account: dict) -> bool:
+        return bool(account.get("api_key"))
+
+    def is_enabled(self, account: dict) -> bool:
+        return account.get("enabled", True)
+
+    def disabled_reason(self, account: dict) -> str:
+        if not account.get("api_key"):
+            return "Neynar API Key is required"
+        if not account.get("signer_uuid"):
+            return "Signer UUID is required for write operations"
+        return ""
+
+    def describe_account(self, account: dict) -> dict:
+        return {
+            "account_id": account.get("account_id", ""),
+            "name": account.get("name", ""),
+            "configured": account.get("configured", False),
+            "fname": account.get("fname", ""),
+        }
+
+    def config_schema(self) -> dict:
+        return {
+            "type": "object",
+            "properties": {
+                "enabled": {"type": "boolean", "default": True, "title": "启用"},
+                "api_key": {
+                    "type": "string",
+                    "title": "Neynar API Key",
+                    "description": "从 dev.neynar.com 获取的 API Key",
+                    "sensitive": True,
+                },
+                "signer_uuid": {
+                    "type": "string",
+                    "title": "Signer UUID",
+                    "description": "Neynar 托管签名的 UUID（写操作必需）",
+                    "sensitive": True,
+                },
+                "fid": {
+                    "type": "integer",
+                    "title": "Bot FID",
+                    "description": "Bot 的 Farcaster FID",
+                },
+                "fname": {
+                    "type": "string",
+                    "title": "Bot Username",
+                    "description": "Bot 的 Farcaster 用户名（用于 @提及匹配）",
+                },
+                "channel_id": {
+                    "type": "string",
+                    "title": "默认频道 ID",
+                    "description": "Farcaster Channel ID，所有公开 Cast 默认发布到此频道",
+                },
+                "dm_policy": {
+                    "type": "string",
+                    "enum": ["pairing", "allowlist", "open", "disabled"],
+                    "default": "pairing",
+                    "title": "DM 安全策略",
+                    "description": "pairing=配对码模式, allowlist=白名单, open=开放, disabled=禁用",
+                },
+                "allow_from": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "title": "允许列表",
+                    "description": 'FID 白名单（如 ["1234", "5678"])',
+                },
+                "webhook_base_url": {
+                    "type": "string",
+                    "title": "Webhook 回调地址",
+                    "description": "Neynar Webhook 推送的公网地址（如 https://forcepilot.example.com）",
+                },
+                "default_account": {"type": "string", "default": "default"},
+                "accounts": {
+                    "type": "object",
+                    "additionalProperties": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "enabled": {"type": "boolean", "default": True},
+                            "api_key": {"type": "string", "sensitive": True},
+                            "signer_uuid": {"type": "string", "sensitive": True},
+                            "fid": {"type": "integer"},
+                            "fname": {"type": "string"},
+                            "channel_id": {"type": "string"},
+                            "dm_policy": {
+                                "type": "string",
+                                "enum": ["pairing", "allowlist", "open", "disabled"],
+                            },
+                            "allow_from": {"type": "array", "items": {"type": "string"}},
+                            "webhook_base_url": {"type": "string"},
+                        },
+                    },
+                },
+            },
+        }
+
+    @staticmethod
+    def _resolve_fc_config(config: dict) -> dict:
+        channels = config.get("channels", {})
+        fc_cfg = channels.get("farcaster", {})
+        return fc_cfg if isinstance(fc_cfg, dict) else {}
+
+    def _get_farcaster_config(self) -> dict:
+        return self._resolve_fc_config(self._config)
+
+
+def _read_file(filepath: str) -> str:
+    if not filepath:
+        return ""
+    try:
+        path = Path(filepath)
+        if path.exists():
+            return path.read_text(encoding="utf-8").strip()
+    except OSError:
+        pass
+    return ""
diff --git a/backend/package/yuxi/channel/extensions/farcaster/dedupe.py b/backend/package/yuxi/channel/extensions/farcaster/dedupe.py
new file mode 100644
index 00000000..601ba7eb
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/dedupe.py
@@ -0,0 +1,30 @@
+import time
+from collections import OrderedDict
+
+from yuxi.channel.extensions.farcaster.defaults import DEDUPE_MAX_TRACKED, DEDUPE_TTL_S
+
+
+class FarcasterDeduplicator:
+    def __init__(self):
+        self._seen: OrderedDict[str, float] = OrderedDict()
+
+    def is_duplicate(self, msg_id: str) -> bool:
+        self._evict_expired()
+        if msg_id in self._seen:
+            return True
+        self._seen[msg_id] = time.time()
+        self._trim()
+        return False
+
+    def reset(self) -> None:
+        self._seen.clear()
+
+    def _evict_expired(self) -> None:
+        now = time.time()
+        expired = [k for k, v in self._seen.items() if now - v > DEDUPE_TTL_S]
+        for key in expired:
+            self._seen.pop(key, None)
+
+    def _trim(self) -> None:
+        while len(self._seen) > DEDUPE_MAX_TRACKED:
+            self._seen.popitem(last=False)
diff --git a/backend/package/yuxi/channel/extensions/farcaster/defaults.py b/backend/package/yuxi/channel/extensions/farcaster/defaults.py
new file mode 100644
index 00000000..9cd94cd7
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/defaults.py
@@ -0,0 +1,18 @@
+CAST_BYTE_LIMIT = 320
+BLOCK_CHUNK_MIN = 240
+BLOCK_CHUNK_MAX = 300
+
+WEBHOOK_BODY_MAX_BYTES = 128 * 1024
+WEBHOOK_BODY_TIMEOUT_S = 5.0
+
+GATEWAY_HEALTH_CHECK_INTERVAL = 30
+GATEWAY_FALLBACK_THRESHOLD_S = 120
+
+DEDUPE_MAX_TRACKED = 10000
+DEDUPE_TTL_S = 600
+
+RETRY_MAX_ATTEMPTS = 3
+RETRY_BASE_DELAY_S = 1.0
+RETRY_MAX_DELAY_S = 10.0
+
+CLIENT_TIMEOUT_S = 30.0
diff --git a/backend/package/yuxi/channel/extensions/farcaster/direct_cast_client.py b/backend/package/yuxi/channel/extensions/farcaster/direct_cast_client.py
new file mode 100644
index 00000000..7f2ea53c
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/direct_cast_client.py
@@ -0,0 +1,57 @@
+import logging
+import uuid
+
+import httpx
+
+WARPCAST_API_BASE = "https://client.warpcast.com/v2"
+
+logger = logging.getLogger(__name__)
+
+
+class DirectCastError(Exception):
+    pass
+
+
+class DirectCastClient:
+    def __init__(self, api_key: str, signer_uuid: str, timeout: float = 30.0):
+        self._api_key = api_key
+        self._signer_uuid = signer_uuid
+        self._client: httpx.AsyncClient | None = None
+        self._timeout = timeout
+
+    async def _ensure_client(self) -> httpx.AsyncClient:
+        if self._client is None:
+            self._client = httpx.AsyncClient(
+                base_url=WARPCAST_API_BASE,
+                headers={
+                    "accept": "application/json",
+                    "x-api-key": self._api_key,
+                },
+                timeout=httpx.Timeout(self._timeout),
+            )
+        return self._client
+
+    async def send_direct_cast(
+        self,
+        recipient_fid: int,
+        text: str,
+        *,
+        embeds: list[dict] | None = None,
+    ) -> dict:
+        client = await self._ensure_client()
+        body: dict = {
+            "recipient_fid": recipient_fid,
+            "text": text,
+            "signer_uuid": self._signer_uuid,
+            "idem": str(uuid.uuid4()),
+        }
+        if embeds:
+            body["embeds"] = embeds
+        resp = await client.post("/ext-send-direct-cast", json=body)
+        resp.raise_for_status()
+        return resp.json()
+
+    async def close(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
diff --git a/backend/package/yuxi/channel/extensions/farcaster/gateway.py b/backend/package/yuxi/channel/extensions/farcaster/gateway.py
new file mode 100644
index 00000000..dbc21a7f
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/gateway.py
@@ -0,0 +1,175 @@
+import asyncio
+import logging
+import time
+
+from yuxi.channel.extensions.farcaster.client import NeynarClient, NeynarError
+from yuxi.channel.extensions.farcaster.defaults import (
+    GATEWAY_FALLBACK_THRESHOLD_S,
+    GATEWAY_HEALTH_CHECK_INTERVAL,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class FarcasterGatewayError(Exception):
+    pass
+
+
+class FarcasterGateway:
+    def __init__(self):
+        self._handles: dict[str, dict] = {}
+
+    async def start(self, ctx) -> dict:
+        account_id = getattr(ctx, "account_id", "default")
+        config = getattr(ctx, "config", {}) or {}
+        channels = config.get("channels", {})
+        fc_cfg = channels.get("farcaster", {}) if isinstance(channels, dict) else {}
+
+        api_key = fc_cfg.get("api_key", "")
+        signer_uuid = fc_cfg.get("signer_uuid", "")
+        webhook_base_url = fc_cfg.get("webhook_base_url", "")
+
+        if not api_key:
+            raise FarcasterGatewayError("api_key not configured")
+
+        client = NeynarClient(api_key=api_key, signer_uuid=signer_uuid)
+
+        try:
+            me = await client.get_me()
+        except NeynarError as e:
+            await client.close()
+            raise FarcasterGatewayError(f"Failed to verify API key: {e}") from e
+
+        bot_fid = me.get("fid", 0)
+        bot_fname = me.get("username", "")
+
+        logger.info(
+            "Farcaster bot verified: fid=%s fname=%s display_name=%s",
+            bot_fid,
+            bot_fname,
+            me.get("display_name", ""),
+        )
+
+        handle = {
+            "client": client,
+            "webhook_id": None,
+            "bot_fid": bot_fid,
+            "bot_fname": bot_fname,
+            "fallback_task": None,
+            "last_webhook_event_at": time.time(),
+            "running": True,
+        }
+
+        if webhook_base_url and bot_fid:
+            webhook_url = f"{webhook_base_url.rstrip('/')}/farcaster/webhook?account_id={account_id}"
+            filters = {
+                "cast.created": {
+                    "author_fids": [bot_fid],
+                    "mentioned_fids": [bot_fid],
+                },
+                "reaction.created": {
+                    "author_fids": [bot_fid],
+                },
+            }
+            try:
+                result = await client.register_webhook(webhook_url, filters)
+                webhook_obj = result.get("webhook", {})
+                handle["webhook_id"] = webhook_obj.get("object") if isinstance(webhook_obj, dict) else None
+                logger.info("Neynar webhook registered: id=%s", handle["webhook_id"])
+            except NeynarError as e:
+                logger.warning("Webhook registration failed, will rely on polling fallback: %s", e)
+
+        handle["fallback_task"] = asyncio.create_task(
+            self._health_check_loop(account_id, handle, ctx),
+            name=f"farcaster-health-{account_id}",
+        )
+
+        self._handles[account_id] = handle
+        return handle
+
+    async def stop(self, ctx) -> None:
+        account_id = getattr(ctx, "account_id", "default")
+        handle = self._handles.pop(account_id, None)
+        if handle is None:
+            return
+
+        handle["running"] = False
+
+        fallback_task = handle.get("fallback_task")
+        if fallback_task:
+            fallback_task.cancel()
+            try:
+                await fallback_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
+        webhook_id = handle.get("webhook_id")
+        client = handle.get("client")
+        if webhook_id and client:
+            try:
+                await client.delete_webhook(webhook_id)
+                logger.info("Neynar webhook deleted: id=%s", webhook_id)
+            except NeynarError as e:
+                logger.warning("Failed to delete webhook: %s", e)
+
+        if client:
+            await client.close()
+        logger.info("Farcaster gateway stopped for account %s", account_id)
+
+    def get_handle(self, account_id: str = "default") -> dict | None:
+        return self._handles.get(account_id)
+
+    @property
+    def account_count(self) -> int:
+        return len(self._handles)
+
+    async def _health_check_loop(self, account_id: str, handle: dict, ctx) -> None:
+        STORAGE_WARN_THRESHOLD = 0.9
+        while handle.get("running", False):
+            await asyncio.sleep(GATEWAY_HEALTH_CHECK_INTERVAL)
+            elapsed = time.time() - handle.get("last_webhook_event_at", 0)
+            if elapsed > GATEWAY_FALLBACK_THRESHOLD_S:
+                logger.warning(
+                    "Farcaster webhook unhealthy for %s (%ss since last event), falling back to polling",
+                    account_id,
+                    int(elapsed),
+                )
+                await self._poll_notifications(handle, ctx)
+
+            client = handle.get("client")
+            bot_fid = handle.get("bot_fid")
+            if client and bot_fid:
+                try:
+                    storage = await client.get_storage_usage(bot_fid)
+                    used = storage.get("used", 0)
+                    limit = storage.get("limit", 0)
+                    if limit > 0 and used / limit > STORAGE_WARN_THRESHOLD:
+                        logger.warning(
+                            "Farcaster storage nearly exhausted: fid=%s used=%s limit=%s ratio=%.1f%%",
+                            bot_fid,
+                            used,
+                            limit,
+                            used / limit * 100,
+                        )
+                except NeynarError:
+                    pass
+
+    async def _poll_notifications(self, handle: dict, ctx) -> None:
+        client = handle.get("client")
+        bot_fid = handle.get("bot_fid")
+        if not client or not bot_fid:
+            return
+
+        try:
+            result = await client.get_notifications(bot_fid, limit=25)
+            notifications = result.get("notifications", [])
+            for notif in notifications:
+                event_type = notif.get("type", "")
+                if event_type in ("cast-reply", "mentions"):
+                    cast = notif.get("cast", {})
+                    if cast:
+                        queue = getattr(ctx, "queue", None)
+                        if queue is not None:
+                            await queue.put({"type": "cast.created", "data": cast})
+        except NeynarError as e:
+            logger.error("Polling fallback failed: %s", e)
diff --git a/backend/package/yuxi/channel/extensions/farcaster/monitor.py b/backend/package/yuxi/channel/extensions/farcaster/monitor.py
new file mode 100644
index 00000000..a5f36369
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/monitor.py
@@ -0,0 +1,132 @@
+import logging
+import re
+from datetime import datetime
+
+from yuxi.channel.message.models import MessageType, PeerInfo, UnifiedMessage
+from yuxi.channel.routing.models import PeerKind
+
+logger = logging.getLogger(__name__)
+
+
+class FarcasterMonitor:
+    def __init__(self, bot_fid: int = 0, bot_fname: str = ""):
+        self._bot_fid = bot_fid
+        self._bot_fname = bot_fname
+
+    def convert_cast_to_unified(
+        self,
+        cast: dict,
+        *,
+        account_id: str = "default",
+    ) -> UnifiedMessage | None:
+        text = cast.get("text", "")
+        author = cast.get("author", {})
+        author_fid = author.get("fid")
+
+        if author_fid is None:
+            logger.warning("Farcaster cast missing author.fid, skipping")
+            return None
+
+        if author_fid == self._bot_fid:
+            return None
+
+        clean_text = _strip_bot_mentions(text, self._bot_fname, self._bot_fid) if self._bot_fname else text
+
+        display_name = author.get("display_name") or author.get("username") or f"fid:{author_fid}"
+
+        timestamp = None
+        ts_str = cast.get("timestamp", "")
+        if ts_str:
+            try:
+                timestamp = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
+            except (ValueError, TypeError):
+                pass
+
+        embeds = cast.get("embeds", [])
+        media_urls = []
+        for embed in embeds:
+            if isinstance(embed, dict) and "url" in embed:
+                media_urls.append(embed["url"])
+
+        return UnifiedMessage(
+            msg_id=_make_msg_id("cast", cast.get("hash", "")),
+            channel_type="farcaster",
+            account_id=account_id,
+            content=clean_text or text,
+            sender=PeerInfo(
+                kind=PeerKind.DIRECT,
+                id=f"farcaster:{author_fid}",
+                display_name=display_name,
+            ),
+            message_type=MessageType.TEXT,
+            media_urls=media_urls if media_urls else None,
+            timestamp=timestamp,
+            reply_to_id=cast.get("parent_hash"),
+            raw_payload=cast,
+            metadata={
+                "cast_hash": cast.get("hash", ""),
+                "thread_hash": cast.get("thread_hash"),
+                "author_fid": author_fid,
+                "author_fname": author.get("username", ""),
+                "pfp_url": author.get("pfp_url"),
+                "channel": cast.get("channel"),
+            },
+        )
+
+    def convert_reaction_to_unified(
+        self,
+        reaction: dict,
+        *,
+        account_id: str = "default",
+    ) -> UnifiedMessage | None:
+        reaction_type = reaction.get("reaction_type", "")
+        user = reaction.get("user", {})
+        cast = reaction.get("cast", {})
+        user_fid = user.get("fid")
+
+        if user_fid is None:
+            return None
+
+        if user_fid == self._bot_fid:
+            return None
+
+        display_name = user.get("display_name") or user.get("username") or f"fid:{user_fid}"
+        cast_hash = cast.get("hash", "")
+
+        return UnifiedMessage(
+            msg_id=_make_msg_id("reaction", f"{cast_hash}:{user_fid}:{reaction_type}"),
+            channel_type="farcaster",
+            account_id=account_id,
+            content=f"[{reaction_type}] cast/{cast_hash}",
+            sender=PeerInfo(
+                kind=PeerKind.DIRECT,
+                id=f"farcaster:{user_fid}",
+                display_name=display_name,
+            ),
+            message_type=MessageType.EVENT,
+            raw_payload=reaction,
+            timestamp=datetime.now(datetime.UTC),
+            metadata={
+                "reaction_type": reaction_type,
+                "cast_hash": cast_hash,
+                "cast_text": cast.get("text", ""),
+                "author_fid": user_fid,
+                "author_fname": user.get("username", ""),
+            },
+        )
+
+    def extract_mentions(self, text: str) -> list[str]:
+        return re.findall(r"@(\w+)", text)
+
+    def strip_mentions(self, text: str) -> str:
+        return re.sub(r"@\w+", "", text).strip()
+
+
+def _strip_bot_mentions(text: str, bot_fname: str, bot_fid: int) -> str:
+    text = re.sub(rf"@?{re.escape(bot_fname)}\b", "", text)
+    text = re.sub(rf"@fid:{bot_fid}", "", text)
+    return text.strip()
+
+
+def _make_msg_id(prefix: str, hash_val: str) -> str:
+    return f"{prefix}:{hash_val}"
diff --git a/backend/package/yuxi/channel/extensions/farcaster/outbound.py b/backend/package/yuxi/channel/extensions/farcaster/outbound.py
new file mode 100644
index 00000000..9f7c97dd
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/outbound.py
@@ -0,0 +1,105 @@
+import logging
+
+from yuxi.channel.extensions.farcaster.client import NeynarClient, NeynarError
+from yuxi.channel.extensions.farcaster.defaults import CAST_BYTE_LIMIT
+from yuxi.channel.protocols import OutboundDeliveryMode
+
+logger = logging.getLogger(__name__)
+
+
+class FarcasterOutboundAdapter:
+    delivery_mode = OutboundDeliveryMode.DIRECT
+    text_chunk_limit = CAST_BYTE_LIMIT
+
+    def __init__(self, client_getter, account_id: str = "default"):
+        self._client_getter = client_getter
+        self._account_id = account_id
+
+    def _get_client(self) -> NeynarClient:
+        client = self._client_getter(self._account_id)
+        if client is None:
+            raise NeynarError("No active Farcaster client")
+        return client
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+        channel_id: str | None = None,
+    ) -> None:
+        text = content
+        if len(text.encode("utf-8")) > CAST_BYTE_LIMIT:
+            text = _trim_to_byte_limit(text, CAST_BYTE_LIMIT)
+
+        parent_hash = reply_to_id or thread_id
+
+        client = self._get_client()
+        try:
+            result = await client.post_cast(
+                text=text,
+                parent_hash=parent_hash,
+                channel_id=channel_id,
+            )
+            cast_hash = result.get("cast", {}).get("hash", "")
+            logger.info(
+                "Farcaster cast sent: hash=%s to=%s parent=%s channel=%s",
+                cast_hash,
+                target_id,
+                parent_hash,
+                channel_id,
+            )
+        except NeynarError as e:
+            logger.error("Farcaster send_text failed: %s", e)
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+        channel_id: str | None = None,
+    ) -> None:
+        captions = {"image": "🖼️ Image", "video": "🎬 Video", "audio": "🎵 Audio", "file": "📎 File"}
+        prefix = captions.get(media_type, media_url)
+
+        text = prefix
+        if len(text.encode("utf-8")) > CAST_BYTE_LIMIT:
+            text = _trim_to_byte_limit(text, CAST_BYTE_LIMIT)
+
+        parent_hash = reply_to_id or thread_id
+
+        client = self._get_client()
+        try:
+            await client.post_cast(
+                text=text,
+                parent_hash=parent_hash,
+                embeds=[{"url": media_url}],
+                channel_id=channel_id,
+            )
+        except NeynarError as e:
+            logger.error("Farcaster send_media failed: type=%s url=%s error=%s", media_type, media_url, e)
+
+    async def send_reaction(
+        self,
+        target_hash: str,
+        reaction_type: str = "like",
+    ) -> None:
+        client = self._get_client()
+        try:
+            await client.post_reaction(target_hash, reaction_type)
+        except NeynarError as e:
+            logger.error("Farcaster send_reaction failed: %s", e)
+
+
+def _trim_to_byte_limit(text: str, limit: int) -> str:
+    encoded = text.encode("utf-8")
+    if len(encoded) <= limit:
+        return text
+    trimmed = encoded[: limit - 3]
+    return trimmed.decode("utf-8", errors="ignore") + "..."
diff --git a/backend/package/yuxi/channel/extensions/farcaster/plugin.json b/backend/package/yuxi/channel/extensions/farcaster/plugin.json
new file mode 100644
index 00000000..248227d6
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/plugin.json
@@ -0,0 +1,28 @@
+{
+  "id": "farcaster",
+  "name": "Farcaster",
+  "version": "1.0.0",
+  "description": "Farcaster decentralized social protocol channel plugin for ForcePilot",
+  "author": "ForcePilot",
+  "order": 105,
+  "enabled": true,
+  "capabilities": {
+    "chat_types": ["direct"],
+    "message_types": ["text"],
+    "interactions": ["recast"],
+    "reactions": true,
+    "typing_indicator": false,
+    "threads": true,
+    "edit": false,
+    "unsend": true,
+    "reply": true,
+    "media": true,
+    "native_commands": false,
+    "polls": false,
+    "streaming": true,
+    "streaming_mode": "block",
+    "block_streaming": true,
+    "block_streaming_chunk_min_chars": 240,
+    "block_streaming_chunk_max_chars": 300
+  }
+}
\ No newline at end of file
diff --git a/backend/package/yuxi/channel/extensions/farcaster/types.py b/backend/package/yuxi/channel/extensions/farcaster/types.py
new file mode 100644
index 00000000..3443c663
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/types.py
@@ -0,0 +1,99 @@
+from pydantic import BaseModel, Field
+
+
+class WebhookEventType:
+    CAST_CREATED = "cast.created"
+    REACTION_CREATED = "reaction.created"
+
+
+class CastAuthor(BaseModel):
+    fid: int
+    username: str | None = None
+    display_name: str | None = None
+    pfp_url: str | None = None
+    follower_count: int = 0
+    following_count: int = 0
+
+
+class CastEmbed(BaseModel):
+    url: str | None = None
+
+
+class CastFrame(BaseModel):
+    url: str | None = None
+
+
+class CastReactionSummary(BaseModel):
+    likes_count: int = 0
+    recasts_count: int = 0
+
+
+class CastData(BaseModel):
+    hash: str
+    thread_hash: str | None = None
+    parent_hash: str | None = None
+    parent_url: str | None = None
+    author: CastAuthor
+    text: str = ""
+    timestamp: str | None = None
+    embeds: list[CastEmbed] = Field(default_factory=list)
+    reactions: CastReactionSummary | None = None
+    replies: dict | None = None
+    mentioned_profiles: list[dict] = Field(default_factory=list)
+    channel: dict | None = None
+
+
+class CastCreatedData(BaseModel):
+    object: str = "cast"
+    hash: str
+    thread_hash: str | None = None
+    parent_hash: str | None = None
+    author: CastAuthor
+    text: str = ""
+    timestamp: str | None = None
+    embeds: list[CastEmbed] = Field(default_factory=list)
+    reactions: CastReactionSummary | None = None
+    replies: dict | None = None
+    mentioned_profiles: list[dict] = Field(default_factory=list)
+    channel: dict | None = None
+
+
+class ReactionUser(BaseModel):
+    fid: int
+    username: str | None = None
+    display_name: str | None = None
+    pfp_url: str | None = None
+
+
+class ReactionCast(BaseModel):
+    hash: str
+    author: CastAuthor | None = None
+    text: str | None = None
+
+
+class ReactionCreatedData(BaseModel):
+    object: str = "reaction"
+    reaction_type: str = "like"
+    cast: ReactionCast
+    user: ReactionUser
+
+
+class WebhookEvent(BaseModel):
+    created_at: int = 0
+    type: str
+    data: dict
+    model_config = {"extra": "allow"}
+
+
+class ResolvedFarcasterAccount(BaseModel):
+    account_id: str
+    name: str = ""
+    enabled: bool = True
+    configured: bool = False
+    api_key: str = ""
+    signer_uuid: str = ""
+    fid: int = 0
+    fname: str = ""
+    dm_policy: str = "pairing"
+    allow_from: list[str] = Field(default_factory=list)
+    webhook_base_url: str = ""
diff --git a/backend/package/yuxi/channel/extensions/farcaster/webhook.py b/backend/package/yuxi/channel/extensions/farcaster/webhook.py
new file mode 100644
index 00000000..1fde62bf
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/farcaster/webhook.py
@@ -0,0 +1,155 @@
+import asyncio
+import json
+import logging
+import time
+from urllib.parse import parse_qs
+
+from fastapi import APIRouter, Header, Request
+from fastapi.responses import PlainTextResponse
+
+from yuxi.channel.extensions.farcaster.defaults import (
+    WEBHOOK_BODY_MAX_BYTES,
+    WEBHOOK_BODY_TIMEOUT_S,
+)
+from yuxi.channel.extensions.farcaster.monitor import FarcasterMonitor
+
+router = APIRouter(prefix="/farcaster/webhook", tags=["farcaster"])
+
+logger = logging.getLogger(__name__)
+
+_GATEWAY_STATE: dict = {}
+
+
+def register_gateway_state(account_id: str, ctx, gateway, dedupe) -> None:
+    handle = gateway.get_handle(account_id)
+    bot_fid = handle.get("bot_fid", 0) if handle else 0
+    bot_fname = handle.get("bot_fname", "") if handle else ""
+
+    monitor = FarcasterMonitor(bot_fid=bot_fid, bot_fname=bot_fname)
+    _GATEWAY_STATE[account_id] = {
+        "ctx": ctx,
+        "gateway": gateway,
+        "monitor": monitor,
+        "dedupe": dedupe,
+    }
+
+
+def unregister_gateway_state(account_id: str) -> None:
+    _GATEWAY_STATE.pop(account_id, None)
+
+
+async def _dispatch_webhook_event(request: Request, account_id: str = "default") -> PlainTextResponse:
+    body = await _read_body(request)
+    if body is None:
+        return PlainTextResponse("", status_code=413)
+
+    x_neynar_signature = request.headers.get("X-Neynar-Signature", "")
+
+    state = _GATEWAY_STATE.get(account_id)
+    if state is None:
+        logger.warning("Farcaster webhook: no active gateway state for account=%s", account_id)
+        return PlainTextResponse("", status_code=503)
+
+    gateway = state["gateway"]
+    handle = gateway.get_handle(account_id)
+    if handle is None or not handle.get("running", False):
+        return PlainTextResponse("", status_code=503)
+
+    client = handle.get("client")
+    if x_neynar_signature and client:
+        if not client.verify_webhook_signature(body, x_neynar_signature):
+            logger.warning("Farcaster webhook: invalid signature")
+            return PlainTextResponse("", status_code=401)
+
+    handle["last_webhook_event_at"] = time.time()
+
+    try:
+        payload = json.loads(body)
+    except json.JSONDecodeError:
+        logger.warning("Farcaster webhook: invalid JSON body")
+        return PlainTextResponse("", status_code=400)
+
+    event_type = payload.get("type", "")
+    event_data = payload.get("data", {})
+
+    ctx = state["ctx"]
+    monitor = state["monitor"]
+    dedupe = state["dedupe"]
+
+    if event_type == "cast.created":
+        await _handle_cast_event(ctx, event_data, handle, monitor, dedupe, account_id)
+    elif event_type == "reaction.created":
+        unified = monitor.convert_reaction_to_unified(event_data, account_id=account_id)
+        if unified is not None and not dedupe.is_duplicate(unified.msg_id):
+            queue = getattr(ctx, "queue", None)
+            if queue is not None:
+                try:
+                    queue.put_nowait(unified)
+                except asyncio.QueueFull:
+                    logger.warning("Farcaster queue full, dropping reaction: %s", unified.msg_id)
+    else:
+        logger.debug("Farcaster unhandled event type: %s", event_type)
+
+    return PlainTextResponse("ok")
+
+
+@router.post("")
+async def farcaster_webhook(
+    request: Request,
+    x_neynar_signature: str = Header("", alias="X-Neynar-Signature"),
+):
+    account_id = "default"
+    parsed = parse_qs(str(request.url.query))
+    if "account_id" in parsed:
+        account_id = parsed["account_id"][0]
+
+    return await _dispatch_webhook_event(request, account_id)
+
+
+async def _handle_cast_event(
+    ctx, cast_data: dict, handle: dict, monitor: FarcasterMonitor, dedupe, account_id: str = "default"
+) -> None:
+    text = cast_data.get("text", "")
+    author = cast_data.get("author", {})
+    bot_fname = handle.get("bot_fname", "")
+    bot_fid = handle.get("bot_fid", 0)
+
+    if not _check_mention(text, bot_fname, bot_fid):
+        logger.debug(
+            "Farcaster cast: not mentioned, fid=%s text=%.100s",
+            author.get("fid", "?"),
+            text,
+        )
+        return
+
+    unified = monitor.convert_cast_to_unified(cast_data, account_id=account_id)
+    if unified is None:
+        return
+
+    if dedupe.is_duplicate(unified.msg_id):
+        return
+
+    queue = getattr(ctx, "queue", None)
+    if queue is not None:
+        try:
+            queue.put_nowait(unified)
+        except asyncio.QueueFull:
+            logger.warning("Farcaster queue full, dropping message: %s", unified.msg_id)
+
+
+def _check_mention(text: str, bot_fname: str, bot_fid: int) -> bool:
+    if not bot_fname:
+        return False
+    return f"@{bot_fname}" in text or f"@fid:{bot_fid}" in text
+
+
+async def _read_body(request: Request) -> bytes | None:
+    try:
+        body = await asyncio.wait_for(request.body(), timeout=WEBHOOK_BODY_TIMEOUT_S)
+    except TimeoutError:
+        logger.warning("Farcaster webhook: body read timeout")
+        return None
+    if len(body) > WEBHOOK_BODY_MAX_BYTES:
+        logger.warning("Farcaster webhook: body too large (%d bytes)", len(body))
+        return None
+    return body