[eric] mcp: TikTok MCP shim (HTTP reads + browser-delegated writes, 11 tools)

backend/apps/tiktok_mcp_shim/__main__.py

@@ -0,0 +1,5 @@
+"""Module-level entrypoint so `python -m backend.apps.tiktok_mcp_shim` works."""
+from backend.apps.tiktok_mcp_shim.server import main
+
+if __name__ == "__main__":
+    main()

backend/apps/tiktok_mcp_shim/handlers.py

@@ -0,0 +1,64 @@
+"""Dispatch each MCP tool call to the TikTok client and format MCP content."""
+
+import json
+from typing import Any, Dict
+
+from backend.apps.social_shims.browser_action import BrowserActionError
+from backend.apps.social_shims.session_source import SessionUnavailable
+from backend.apps.tiktok_mcp_shim import tiktok_reads as reads
+from backend.apps.tiktok_mcp_shim import tiktok_writes as writes
+from backend.apps.tiktok_mcp_shim.tiktok_http import TikTokError
+
+
+def mcp_ok(payload: Any) -> Dict[str, Any]:
+    if isinstance(payload, str):
+        return {"content": [{"type": "text", "text": payload}]}
+    return {"content": [{"type": "text", "text": json.dumps(payload, indent=2, default=str)}]}
+
+
+def mcp_err(text: str) -> Dict[str, Any]:
+    return {"content": [{"type": "text", "text": f"Error: {text}"}], "isError": True}
+
+
+def handle_tool_call(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
+    try:
+        return mcp_ok(p_dispatch(name, args))
+    except SessionUnavailable as e:
+        return mcp_err(str(e))
+    except (TikTokError, BrowserActionError) as e:
+        return mcp_err(str(e))
+    except Exception as e:
+        return mcp_err(f"tiktok shim error: {e!r}")
+
+
+def p_dispatch(name: str, a: Dict[str, Any]) -> Any:
+    if name == "tiktok_feed":
+        return reads.feed(p_lim(a.get("count"), 20))
+    if name == "tiktok_search":
+        return reads.search(a.get("keyword", ""), p_lim(a.get("count"), 20))
+    if name == "tiktok_get_user":
+        return reads.get_user(a.get("username", ""))
+    if name == "tiktok_user_videos":
+        return reads.user_videos(a.get("username", ""), p_lim(a.get("count"), 20), a.get("cursor", ""))
+    if name == "tiktok_get_video":
+        return reads.get_video(a.get("video_id", ""))
+    if name == "tiktok_comments":
+        return reads.comments(a.get("video_id", ""), p_lim(a.get("count"), 20), a.get("cursor", ""))
+    if name == "tiktok_like":
+        return writes.like(a.get("video_url", ""), bool(a.get("unlike")))
+    if name == "tiktok_favorite":
+        return writes.favorite(a.get("video_url", ""), bool(a.get("remove")))
+    if name == "tiktok_comment":
+        return writes.comment(a.get("video_url", ""), a.get("text", ""))
+    if name == "tiktok_follow":
+        return writes.follow(a.get("username", ""), bool(a.get("unfollow")))
+    if name == "tiktok_upload":
+        return writes.upload(a.get("caption", ""), a.get("video_path", ""))
+    raise TikTokError(f"Unknown tool: {name}")
+
+
+def p_lim(v: Any, default: int) -> int:
+    try:
+        return max(1, min(int(v), 50))
+    except (TypeError, ValueError):
+        return default

backend/apps/tiktok_mcp_shim/rate_limit.py

@@ -0,0 +1,32 @@
+"""TikTok's per-action pacing config on top of the shared RateLimiter.
+
+Reads are generous; likes/favorites moderate, comments/follows slow, so the account
+never bursts like a bot. The shared core owns the algorithm + 429 backoff.
+"""
+
+from typing import Dict, Tuple
+
+from backend.apps.social_shims.rate_limit_core import RateLimiter
+
+# action -> (bucket_capacity, seconds_to_refill_one_token).
+BUCKETS: Dict[str, Tuple[float, float]] = {
+    "read": (25.0, 1.0),
+    "like": (15.0, 3.0),
+    "favorite": (15.0, 3.0),
+    "comment": (5.0, 15.0),
+    "follow": (8.0, 8.0),
+}
+
+p_limiter = RateLimiter(BUCKETS, min_gap_s=1.2, jitter_s=0.8)
+
+
+def acquire(action: str) -> None:
+    p_limiter.acquire(action)
+
+
+def note_response(status: int, headers: Dict[str, str]) -> None:
+    p_limiter.note_response(status, headers)
+
+
+def bucket_for(action: str) -> str:
+    return p_limiter.bucket_for(action)

backend/apps/tiktok_mcp_shim/server.py

@@ -0,0 +1,63 @@
+"""Stdio JSON-RPC MCP server for TikTok.
+
+Mirrors the discord/reddit/x shim loop: stdlib-only, no backend.config imports, so the
+subprocess starts fast. The tool surface lives in tools.py; dispatch in handlers.py.
+"""
+
+import json
+import sys
+from typing import Any, Optional
+
+from backend.apps.tiktok_mcp_shim.handlers import handle_tool_call, mcp_err
+from backend.apps.tiktok_mcp_shim.tools import TOOLS
+
+
+def p_send(id_: Any, result: Optional[dict] = None, error: Optional[dict] = None) -> None:
+    msg: dict = {"jsonrpc": "2.0", "id": id_}
+    if error is not None:
+        msg["error"] = error
+    else:
+        msg["result"] = result
+    sys.stdout.write(json.dumps(msg) + "\n")
+    sys.stdout.flush()
+
+
+def main() -> None:
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            msg = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+
+        method = msg.get("method")
+        id_ = msg.get("id")
+        params = msg.get("params", {}) or {}
+
+        if method == "initialize":
+            p_send(id_, {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {"tools": {}},
+                "serverInfo": {"name": "openswarm-tiktok", "version": "1.0.0"},
+            })
+        elif method == "notifications/initialized":
+            pass
+        elif method == "tools/list":
+            p_send(id_, {"tools": TOOLS})
+        elif method == "tools/call":
+            name = params.get("name", "")
+            args = params.get("arguments", {}) or {}
+            try:
+                p_send(id_, handle_tool_call(name, args))
+            except Exception as e:
+                p_send(id_, mcp_err(f"shim crashed: {e!r}"))
+        elif method == "ping":
+            p_send(id_, {})
+        elif id_ is not None:
+            p_send(id_, error={"code": -32601, "message": f"Method not found: {method}"})
+
+
+if __name__ == "__main__":
+    main()

backend/apps/tiktok_mcp_shim/tiktok_endpoints.py

@@ -0,0 +1,37 @@
+"""TikTok web-client constants + device params, and the request-signing reality.
+
+TikTok gates almost every /api call behind a per-request signature: an msToken in the
+cookies PLUS an X-Bogus / X-Gnarly query param generated by obfuscated client JS. We
+attach the borrowed msToken (some read endpoints accept it), but TikTok rotates the
+X-Bogus/X-Gnarly scheme constantly and generating it outside a real browser is the
+known hard wall for every free TikTok client. So reads here are best-effort, and signed
+writes (and video upload) realistically need OpenSwarm's in-app browser agent, which
+drives the real logged-in session and is therefore free AND undetectable AND able to do
+everything a human can. This module isolates the brittle bits; see tiktok_sign.py.
+"""
+
+API = "https://www.tiktok.com/api"
+DOMAIN = "tiktok.com"
+MSTOKEN_COOKIE = "msToken"
+
+# Static-ish web/device params TikTok expects on every /api call.
+DEVICE_PARAMS = {
+    "aid": "1988",
+    "app_name": "tiktok_web",
+    "app_language": "en",
+    "channel": "tiktok_web",
+    "device_platform": "web_pc",
+    "cookie_enabled": "true",
+    "screen_width": "1512",
+    "screen_height": "982",
+    "browser_language": "en-US",
+    "browser_platform": "MacIntel",
+    "browser_name": "Mozilla",
+    "browser_version": "5.0",
+    "browser_online": "true",
+    "timezone_name": "America/Los_Angeles",
+    "os": "mac",
+    "region": "US",
+    "priority_region": "US",
+    "language": "en",
+}

backend/apps/tiktok_mcp_shim/tiktok_http.py

@@ -0,0 +1,89 @@
+"""Low-level authed TikTok transport.
+
+Borrow the user's tiktok.com session, attach device params + msToken, and call the
+web /api surface. Detects TikTok's anti-bot / verify rejections (the signature gate)
+and raises an actionable error pointing at the browser fallback. stdlib-only.
+"""
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any, Dict, Optional
+
+from backend.apps.social_shims.session_source import get_session, invalidate
+from backend.apps.tiktok_mcp_shim import rate_limit
+from backend.apps.tiktok_mcp_shim.tiktok_endpoints import API, DOMAIN
+from backend.apps.tiktok_mcp_shim.tiktok_sign import signed_query
+
+SIGNATURE_HINT = (
+    "TikTok blocked this as unsigned/automated (its X-Bogus/X-Gnarly gate). Reads sometimes "
+    "slip through; signed writes and uploads need a real browser. Use the OpenSwarm browser "
+    "agent for TikTok actions: it drives your live tiktok.com session, so it's free, "
+    "undetectable, and can do everything a human can."
+)
+
+
+class TikTokError(Exception):
+    """A TikTok request failed in a way worth surfacing to the agent."""
+
+
+def check_antibot(body: Any) -> None:
+    if not isinstance(body, dict):
+        return
+    sc = body.get("statusCode", body.get("status_code"))
+    if sc not in (0, None):
+        msg = body.get("statusMsg") or body.get("status_msg") or ""
+        raise TikTokError(f"TikTok statusCode {sc} {msg}. {SIGNATURE_HINT}".strip())
+
+
+def p_request(method: str, path: str, *, params: Optional[Dict[str, Any]],
+              form: Optional[Dict[str, Any]], action: str, retried: bool) -> Any:
+    rate_limit.acquire(action)
+    cookie, ua = get_session(DOMAIN)
+    url = f"{API}/{path}?" + signed_query(params or {})
+    data = None
+    if form is not None:
+        data = urllib.parse.urlencode({k: v for k, v in form.items() if v is not None}).encode()
+    headers = {
+        "Cookie": cookie,
+        "User-Agent": ua,
+        "Accept": "application/json, text/plain, */*",
+        "Referer": "https://www.tiktok.com/",
+    }
+    if data is not None:
+        headers["Content-Type"] = "application/x-www-form-urlencoded"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    try:
+        with urllib.request.urlopen(req, timeout=30.0) as resp:
+            status, raw, rhdr = resp.status, resp.read(), dict(resp.headers)
+    except urllib.error.HTTPError as e:
+        status, raw, rhdr = e.code, (e.read() if e.fp else b""), dict(e.headers or {})
+    except urllib.error.URLError as e:
+        raise TikTokError(f"tiktok.com unreachable: {getattr(e, 'reason', e)}")
+
+    rate_limit.note_response(status, {k.lower(): v for k, v in rhdr.items()})
+    if status in (401, 403) and not retried:
+        invalidate(DOMAIN)
+        return p_request(method, path, params=params, form=form, action=action, retried=True)
+    if status == 429:
+        raise TikTokError("tiktok.com is rate-limiting this account; slow down and retry shortly.")
+    if status >= 400:
+        raise TikTokError(f"tiktok.com HTTP {status}: {raw[:200].decode('utf-8', 'replace')}. {SIGNATURE_HINT}")
+    text = raw.decode("utf-8", errors="replace").strip()
+    if not text:
+        raise TikTokError(f"TikTok returned an empty response. {SIGNATURE_HINT}")
+    try:
+        body = json.loads(text)
+    except json.JSONDecodeError:
+        raise TikTokError(f"TikTok returned a non-JSON page (likely a verify/captcha wall). {SIGNATURE_HINT}")
+    check_antibot(body)
+    return body
+
+
+def get(path: str, params: Dict[str, Any], *, action: str = "read") -> Any:
+    return p_request("GET", path, params=params, form=None, action=action, retried=False)
+
+
+def post(path: str, params: Dict[str, Any], form: Dict[str, Any], *, action: str) -> Any:
+    return p_request("POST", path, params=params, form=form, action=action, retried=False)

backend/apps/tiktok_mcp_shim/tiktok_reads.py

@@ -0,0 +1,122 @@
+"""Read operations over tiktok.com's web /api surface.
+
+Returns compact video/user/comment records (truncated captions). A defensive walker
+pulls video items out of whatever envelope the endpoint uses (itemList, data[].item),
+which survives TikTok's frequent response reshuffles. Reads are best-effort: TikTok's
+signature gate may still reject them, in which case tiktok_http raises an actionable hint.
+"""
+
+from typing import Any, Dict, List, Optional
+
+from backend.apps.tiktok_mcp_shim.tiktok_http import TikTokError, get
+
+CAP = 800
+
+
+def p_trunc(s: Optional[str]) -> str:
+    s = s or ""
+    return s if len(s) <= CAP else s[:CAP] + f"... [+{len(s) - CAP} chars]"
+
+
+def p_item(d: Dict[str, Any]) -> Dict[str, Any]:
+    author = d.get("author") or {}
+    if not isinstance(author, dict):
+        author = {"uniqueId": author}
+    stats = d.get("stats") or d.get("statsV2") or {}
+    vid = d.get("id") or d.get("aweme_id")
+    handle = author.get("uniqueId")
+    return {
+        "id": vid,
+        "desc": p_trunc(d.get("desc")),
+        "author": handle,
+        "author_name": author.get("nickname"),
+        "likes": stats.get("diggCount"),
+        "comments": stats.get("commentCount"),
+        "plays": stats.get("playCount"),
+        "shares": stats.get("shareCount"),
+        "created": d.get("createTime"),
+        "url": f"https://www.tiktok.com/@{handle}/video/{vid}" if handle and vid else None,
+    }
+
+
+def p_is_item(d: Any) -> bool:
+    return isinstance(d, dict) and "desc" in d and "author" in d and ("id" in d or "aweme_id" in d)
+
+
+def p_collect_items(node: Any, out: List[Dict[str, Any]], cap: int) -> None:
+    if len(out) >= cap:
+        return
+    if p_is_item(node):
+        t = p_item(node)
+        if t.get("id") and not any(x["id"] == t["id"] for x in out):
+            out.append(t)
+        return
+    if isinstance(node, dict):
+        for v in node.values():
+            p_collect_items(v, out, cap)
+    elif isinstance(node, list):
+        for v in node:
+            p_collect_items(v, out, cap)
+
+
+def p_items_out(resp: Any, cap: int) -> Dict[str, Any]:
+    out: List[Dict[str, Any]] = []
+    p_collect_items(resp, out, cap)
+    cursor = resp.get("cursor") if isinstance(resp, dict) else None
+    return {"videos": out, "cursor": cursor, "has_more": bool(resp.get("hasMore")) if isinstance(resp, dict) else None}
+
+
+def get_user(username: str) -> Dict[str, Any]:
+    resp = get("user/detail/", {"uniqueId": username.lstrip("@")})
+    info = (resp or {}).get("userInfo", {})
+    user = info.get("user", {})
+    stats = info.get("stats", {})
+    return {
+        "id": user.get("id"),
+        "sec_uid": user.get("secUid"),
+        "username": user.get("uniqueId"),
+        "nickname": user.get("nickname"),
+        "bio": p_trunc(user.get("signature")),
+        "followers": stats.get("followerCount"),
+        "following": stats.get("followingCount"),
+        "likes": stats.get("heartCount"),
+        "videos": stats.get("videoCount"),
+        "verified": user.get("verified"),
+    }
+
+
+def feed(count: int) -> Dict[str, Any]:
+    return p_items_out(get("recommend/item_list/", {"count": count, "from_page": "fyp"}), count)
+
+
+def user_videos(username: str, count: int, cursor: str) -> Dict[str, Any]:
+    sec_uid = get_user(username).get("sec_uid")
+    if not sec_uid:
+        raise TikTokError(f"Could not resolve @{username.lstrip('@')} to a secUid.")
+    return p_items_out(get("post/item_list/", {"secUid": sec_uid, "count": count, "cursor": cursor or "0"}), count)
+
+
+def get_video(video_id: str) -> Dict[str, Any]:
+    resp = get("item/detail/", {"itemId": video_id})
+    item = (resp or {}).get("itemInfo", {}).get("itemStruct", {})
+    return p_item(item) if item else {"id": video_id, "note": "not found or signature-gated"}
+
+
+def comments(video_id: str, count: int, cursor: str) -> Dict[str, Any]:
+    resp = get("comment/list/", {"aweme_id": video_id, "count": count, "cursor": cursor or "0"})
+    out = []
+    for c in (resp or {}).get("comments", []) or []:
+        u = c.get("user", {})
+        out.append({
+            "id": c.get("cid"),
+            "text": p_trunc(c.get("text")),
+            "author": u.get("unique_id") or u.get("uniqueId"),
+            "likes": c.get("digg_count"),
+            "created": c.get("create_time"),
+        })
+    return {"comments": out, "cursor": resp.get("cursor") if isinstance(resp, dict) else None}
+
+
+def search(keyword: str, count: int) -> Dict[str, Any]:
+    resp = get("search/general/full/", {"keyword": keyword, "offset": 0, "count": count, "from_page": "search"})
+    return p_items_out(resp, count)

backend/apps/tiktok_mcp_shim/tiktok_sign.py

@@ -0,0 +1,27 @@
+"""Attach TikTok's request signature.
+
+msToken is borrowed from the user's session cookies and appended to the query, which
+some read endpoints accept. X-Bogus / X-Gnarly is TikTok's rotating anti-automation
+signature, generated by obfuscated browser JS; we deliberately do NOT reproduce it
+here, because a stale or wrong signature is worse than none and TikTok changes it on a
+whim. When a call comes back signature-rejected, the shim says so and points at the
+browser agent, which signs naturally by being a real browser. This is the TikTok analog
+of Reddit's token-harvest and X's queryIds: the one isolated, not-live-proven piece.
+"""
+
+import urllib.parse
+from typing import Any, Dict
+
+from backend.apps.social_shims.session_source import cookie_value
+from backend.apps.tiktok_mcp_shim.tiktok_endpoints import DEVICE_PARAMS, DOMAIN, MSTOKEN_COOKIE
+
+
+def signed_query(params: Dict[str, Any]) -> str:
+    """Build the query string with device params + the borrowed msToken."""
+    q = dict(DEVICE_PARAMS)
+    q.update({k: v for k, v in params.items() if v is not None})
+    ms = cookie_value(DOMAIN, MSTOKEN_COOKIE)
+    if ms:
+        q["msToken"] = ms
+    # X-Bogus / X-Gnarly intentionally omitted; see module docstring.
+    return urllib.parse.urlencode(q)

backend/apps/tiktok_mcp_shim/tiktok_writes.py

@@ -0,0 +1,91 @@
+"""Write operations for TikTok, delegated to the user's own live browser card.
+
+TikTok signs every request, so pure-HTTP writes get bot-flagged. Instead each write drives
+the user's already-open, logged-in tiktok.com card via the backend action bridge: navigate to
+the target, then run a small click/type script keyed on TikTok's data-e2e test-ids (with a
+button-text fallback). The card is a real signed-in browser, so this is free, undetectable, and
+does what a human does. Selector drift is the isolated soft spot: if a control moves, the script
+returns an actionable error and the agent can fall back to driving the card by hand.
+"""
+
+import json
+from typing import Any, Dict, List
+
+from backend.apps.social_shims.browser_action import last_json, perform
+
+DOMAIN = "tiktok.com"
+UPLOAD_URL = "https://www.tiktok.com/upload"
+
+
+def p_click_script(candidates: List[str], label: str) -> str:
+    """JS that polls up to ~6s for the first matching control (by selector, then button text) and clicks it."""
+    cands = json.dumps(candidates)
+    lbl = json.dumps(label)
+    return (
+        "(async()=>{const cands=" + cands + ";const label=" + lbl + ";"
+        "const find=()=>{for(const c of cands){const el=document.querySelector(c);if(el)return el;}"
+        "for(const b of document.querySelectorAll('button,[role=button],[data-e2e]')){"
+        "if((b.textContent||'').trim().toLowerCase()===label.toLowerCase())return b;}return null;};"
+        "const deadline=Date.now()+6000;let el=find();"
+        "while(!el&&Date.now()<deadline){await new Promise(r=>setTimeout(r,300));el=find();}"
+        "if(!el)return{ok:false,error:'control not found: '+label};"
+        "el.scrollIntoView({block:'center'});el.click();return{ok:true,clicked:label};})()"
+    )
+
+
+def p_comment_script(text: str) -> str:
+    t = json.dumps(text)
+    return (
+        "(async()=>{const q=s=>document.querySelector(s);const deadline=Date.now()+6000;"
+        "let box=q('[data-e2e=\"comment-input\"]')||q('div[contenteditable=\"true\"]');"
+        "while(!box&&Date.now()<deadline){await new Promise(r=>setTimeout(r,300));"
+        "box=q('[data-e2e=\"comment-input\"]')||q('div[contenteditable=\"true\"]');}"
+        "if(!box)return{ok:false,error:'comment box not found'};"
+        "box.focus();document.execCommand('selectAll',false);document.execCommand('insertText',false," + t + ");"
+        "box.dispatchEvent(new InputEvent('input',{bubbles:true,inputType:'insertText',data:" + t + "}));"
+        "await new Promise(r=>setTimeout(r,400));"
+        "const post=q('[data-e2e=\"comment-post\"]')||q('[data-e2e=\"comment-post-button\"]');"
+        "if(!post)return{ok:false,error:'post button not found'};"
+        "if(post.getAttribute('aria-disabled')==='true')return{ok:false,error:'post button disabled (comment empty?)'};"
+        "post.click();return{ok:true,posted:true};})()"
+    )
+
+
+def p_do(url: str, script: str) -> Dict[str, Any]:
+    steps = [{"op": "navigate", "url": url}, {"op": "wait", "ms": 1500}, {"op": "evaluate", "expression": script}]
+    return last_json(perform(DOMAIN, steps))
+
+
+def like(video_url: str, unlike: bool) -> Dict[str, Any]:
+    out = p_do(video_url, p_click_script(['[data-e2e="like-icon"]', '[data-e2e="browse-like-icon"]'], "like"))
+    return {"video": video_url, "liked": bool(out.get("ok")) and not unlike, "detail": out}
+
+
+def favorite(video_url: str, remove: bool) -> Dict[str, Any]:
+    out = p_do(video_url, p_click_script(['[data-e2e="favorite-icon"]', '[data-e2e="browse-favorite-icon"]'], "favorite"))
+    return {"video": video_url, "favorited": bool(out.get("ok")) and not remove, "detail": out}
+
+
+def follow(username: str, unfollow: bool) -> Dict[str, Any]:
+    handle = username.lstrip("@")
+    label = "following" if unfollow else "follow"
+    out = p_do(f"https://www.tiktok.com/@{handle}", p_click_script(['[data-e2e="follow-button"]', '[data-e2e="follow-icon"]'], label))
+    return {"username": handle, "following": bool(out.get("ok")) and not unfollow, "detail": out}
+
+
+def comment(video_url: str, text: str) -> Dict[str, Any]:
+    out = p_do(video_url, p_comment_script(text))
+    return {"video": video_url, "posted": bool(out.get("ok")), "detail": out}
+
+
+def upload(caption: str, video_path: str) -> Dict[str, Any]:
+    # Open the real upload page; the OS file picker can't be driven from page JS (browser security), so the human/agent finishes the file choice.
+    perform(DOMAIN, [{"op": "navigate", "url": UPLOAD_URL}])
+    return {
+        "opened": UPLOAD_URL,
+        "note": (
+            f"Opened the TikTok upload page in your browser card. Choose the file ({video_path!r}) in the "
+            f"picker and paste the caption ({caption!r}); browser security blocks scripts from selecting the "
+            "file for you, so this last step is yours (or drive the card with the browser agent)."
+        ),
+    }

backend/apps/tiktok_mcp_shim/tools.py

@@ -0,0 +1,133 @@
+"""MCP tool surface for TikTok: the things a logged-in human does.
+
+Reads (feed/search/user/videos/video/comments) and writes (like/favorite/comment/
+follow/upload). Writes are signature-gated by TikTok; when its anti-bot check rejects
+a call the tool returns an actionable error pointing at the OpenSwarm browser agent,
+which drives the real session and is free + undetectable. Video ids come from the read
+tools (or a tiktok.com/@user/video/<id> URL's trailing id).
+"""
+
+OBJ = "object"
+
+TOOLS = [
+    {
+        "name": "tiktok_feed",
+        "description": "Read the For You feed (recommended videos).",
+        "inputSchema": {"type": OBJ, "properties": {"count": {"type": "integer", "default": 20}}},
+    },
+    {
+        "name": "tiktok_search",
+        "description": "Search TikTok videos by keyword.",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "keyword": {"type": "string"},
+                "count": {"type": "integer", "default": 20},
+            },
+            "required": ["keyword"],
+        },
+    },
+    {
+        "name": "tiktok_get_user",
+        "description": "Get a creator's profile (bio, follower/like/video counts) by @handle.",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {"username": {"type": "string"}},
+            "required": ["username"],
+        },
+    },
+    {
+        "name": "tiktok_user_videos",
+        "description": "List a creator's recent videos by @handle.",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "username": {"type": "string"},
+                "count": {"type": "integer", "default": 20},
+                "cursor": {"type": "string"},
+            },
+            "required": ["username"],
+        },
+    },
+    {
+        "name": "tiktok_get_video",
+        "description": "Get a single video's details by its numeric id.",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {"video_id": {"type": "string"}},
+            "required": ["video_id"],
+        },
+    },
+    {
+        "name": "tiktok_comments",
+        "description": "Read the comments on a video by its numeric id.",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "video_id": {"type": "string"},
+                "count": {"type": "integer", "default": 20},
+                "cursor": {"type": "string"},
+            },
+            "required": ["video_id"],
+        },
+    },
+    {
+        "name": "tiktok_like",
+        "description": "Like a video by its URL (the 'url' from a read result; or unlike with unlike=true).",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "video_url": {"type": "string", "description": "Full tiktok.com video URL from a read result."},
+                "unlike": {"type": "boolean", "default": False},
+            },
+            "required": ["video_url"],
+        },
+    },
+    {
+        "name": "tiktok_favorite",
+        "description": "Add a video to your favorites by its URL (or remove with remove=true).",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "video_url": {"type": "string", "description": "Full tiktok.com video URL from a read result."},
+                "remove": {"type": "boolean", "default": False},
+            },
+            "required": ["video_url"],
+        },
+    },
+    {
+        "name": "tiktok_comment",
+        "description": "Post a comment on a video by its URL.",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "video_url": {"type": "string", "description": "Full tiktok.com video URL from a read result."},
+                "text": {"type": "string"},
+            },
+            "required": ["video_url", "text"],
+        },
+    },
+    {
+        "name": "tiktok_follow",
+        "description": "Follow a creator by @handle (or unfollow with unfollow=true).",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "username": {"type": "string"},
+                "unfollow": {"type": "boolean", "default": False},
+            },
+            "required": ["username"],
+        },
+    },
+    {
+        "name": "tiktok_upload",
+        "description": "Upload a video. Routes to the OpenSwarm browser agent (TikTok upload can't be done bot-safely over HTTP).",
+        "inputSchema": {
+            "type": OBJ,
+            "properties": {
+                "caption": {"type": "string"},
+                "video_path": {"type": "string"},
+            },
+        },
+    },
+]

backend/tests/test_tiktok_mcp_shim.py

@@ -0,0 +1,140 @@
+"""Unit coverage for the TikTok MCP shim: query signing (device params + borrowed msToken),
+the anti-bot/verify detector, HTTP read dispatch (item walker + normalizer), and the
+browser-delegated writes (navigate + evaluate the user's own live card). Network + the
+browser bridge are mocked; the live browser round-trip needs the running app + a logged-in
+tiktok.com card, so it's asserted at the step-shape level here."""
+
+import json
+import time
+
+from unittest.mock import patch
+
+from backend.apps.social_shims.browser_action import BrowserActionError
+from backend.apps.social_shims.session_source import SessionUnavailable
+from backend.apps.tiktok_mcp_shim import rate_limit, tiktok_reads, tiktok_sign, tiktok_writes
+from backend.apps.tiktok_mcp_shim.handlers import handle_tool_call
+from backend.apps.tiktok_mcp_shim.tiktok_http import TikTokError, check_antibot
+
+VIDEO_URL = "https://www.tiktok.com/@bob/video/777"
+
+
+def p_text(result: dict) -> str:
+    return result["content"][0]["text"]
+
+
+CANNED_ITEM = {
+    "id": "777", "desc": "a dance",
+    "author": {"uniqueId": "bob", "nickname": "Bob"},
+    "stats": {"diggCount": 9, "commentCount": 2, "playCount": 100, "shareCount": 1},
+    "createTime": 123,
+}
+
+
+# -- signing ---------------------------------------------------------------
+
+def test_signed_query_has_device_params_and_mstoken():
+    with patch.object(tiktok_sign, "cookie_value", return_value="MSTOK123"):
+        q = tiktok_sign.signed_query({"count": 5})
+    assert "aid=1988" in q and "app_name=tiktok_web" in q
+    assert "count=5" in q and "msToken=MSTOK123" in q
+
+
+def test_signed_query_omits_mstoken_when_absent():
+    with patch.object(tiktok_sign, "cookie_value", return_value=""):
+        q = tiktok_sign.signed_query({})
+    assert "msToken=" not in q
+
+
+# -- anti-bot guard --------------------------------------------------------
+
+def test_antibot_raises_with_browser_hint():
+    check_antibot({"statusCode": 0})  # ok, no raise
+    try:
+        check_antibot({"statusCode": 10201, "statusMsg": "verify"})
+        assert False, "expected TikTokError"
+    except TikTokError as e:
+        assert "browser" in str(e).lower()
+
+
+# -- rate limiter ----------------------------------------------------------
+
+def test_first_read_is_prompt():
+    start = time.time()
+    rate_limit.acquire("read")
+    assert time.time() - start < 1.4
+
+
+# -- read dispatch (network mocked; exercises the walker + normalizer) ------
+
+def test_feed_walks_and_normalizes():
+    canned = {"itemList": [CANNED_ITEM], "cursor": "5", "hasMore": True}
+    with patch.object(tiktok_reads, "get", return_value=canned):
+        out = handle_tool_call("tiktok_feed", {"count": 10})
+    data = json.loads(p_text(out))
+    v = data["videos"][0]
+    assert v["id"] == "777" and v["author"] == "bob" and v["likes"] == 9
+    assert v["url"] == VIDEO_URL and data["cursor"] == "5"
+
+
+def test_read_session_unavailable_is_actionable():
+    def boom(*a, **k):
+        raise SessionUnavailable("Not logged in to tiktok.com. Open tiktok.com in the OpenSwarm browser, sign in, then retry.")
+
+    with patch.object(tiktok_reads, "get", boom):
+        out = handle_tool_call("tiktok_get_user", {"username": "bob"})
+    assert out.get("isError") is True and "logged in" in p_text(out).lower()
+
+
+# -- browser-delegated writes (bridge mocked) ------------------------------
+
+def test_like_delegates_navigate_then_evaluate():
+    captured: dict = {}
+
+    def fake_perform(domain, steps):
+        captured["domain"] = domain
+        captured["ops"] = [s["op"] for s in steps]
+        captured["nav"] = steps[0].get("url")
+        return {"ok": True, "results": [{"text": json.dumps({"ok": True, "clicked": "like"})}]}
+
+    with patch.object(tiktok_writes, "perform", fake_perform):
+        out = handle_tool_call("tiktok_like", {"video_url": VIDEO_URL})
+    assert captured["domain"] == "tiktok.com"
+    assert captured["ops"] == ["navigate", "wait", "evaluate"]
+    assert captured["nav"] == VIDEO_URL
+    assert json.loads(p_text(out))["liked"] is True
+
+
+def test_follow_navigates_to_profile():
+    captured: dict = {}
+    with patch.object(tiktok_writes, "perform",
+                      lambda d, steps: captured.update(nav=steps[0]["url"]) or {"results": [{"text": '{"ok":true}'}]}):
+        handle_tool_call("tiktok_follow", {"username": "@bob"})
+    assert captured["nav"] == "https://www.tiktok.com/@bob"
+
+
+def test_write_surfaces_no_card_error():
+    def boom(domain, steps):
+        raise BrowserActionError("No tiktok.com browser card is open. Open tiktok.com in an OpenSwarm browser card and sign in, then retry.")
+
+    with patch.object(tiktok_writes, "perform", boom):
+        out = handle_tool_call("tiktok_like", {"video_url": VIDEO_URL})
+    assert out.get("isError") is True and "browser card" in p_text(out).lower()
+
+
+def test_upload_opens_upload_page():
+    captured: dict = {}
+
+    def fake_perform(domain, steps):
+        captured["url"] = steps[0].get("url")
+        return {"ok": True, "results": []}
+
+    with patch.object(tiktok_writes, "perform", fake_perform):
+        out = handle_tool_call("tiktok_upload", {"caption": "hi", "video_path": "/tmp/v.mp4"})
+    data = json.loads(p_text(out))
+    assert data["opened"].endswith("/upload") and captured["url"].endswith("/upload")
+    assert "/tmp/v.mp4" in data["note"]
+
+
+def test_unknown_tool_errors():
+    out = handle_tool_call("tiktok_nonsense", {})
+    assert out.get("isError") is True