diff --git a/backend/apps/tiktok_mcp_shim/__init__.py b/backend/apps/tiktok_mcp_shim/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/apps/tiktok_mcp_shim/__main__.py b/backend/apps/tiktok_mcp_shim/__main__.py new file mode 100644 index 00000000..2cf7de5e --- /dev/null +++ b/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() diff --git a/backend/apps/tiktok_mcp_shim/handlers.py b/backend/apps/tiktok_mcp_shim/handlers.py new file mode 100644 index 00000000..38514461 --- /dev/null +++ b/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 diff --git a/backend/apps/tiktok_mcp_shim/rate_limit.py b/backend/apps/tiktok_mcp_shim/rate_limit.py new file mode 100644 index 00000000..cc5dd55e --- /dev/null +++ b/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) diff --git a/backend/apps/tiktok_mcp_shim/server.py b/backend/apps/tiktok_mcp_shim/server.py new file mode 100644 index 00000000..eafaa68f --- /dev/null +++ b/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() diff --git a/backend/apps/tiktok_mcp_shim/tiktok_endpoints.py b/backend/apps/tiktok_mcp_shim/tiktok_endpoints.py new file mode 100644 index 00000000..97216881 --- /dev/null +++ b/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", +} diff --git a/backend/apps/tiktok_mcp_shim/tiktok_http.py b/backend/apps/tiktok_mcp_shim/tiktok_http.py new file mode 100644 index 00000000..466e7a43 --- /dev/null +++ b/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) diff --git a/backend/apps/tiktok_mcp_shim/tiktok_reads.py b/backend/apps/tiktok_mcp_shim/tiktok_reads.py new file mode 100644 index 00000000..f95000e9 --- /dev/null +++ b/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) diff --git a/backend/apps/tiktok_mcp_shim/tiktok_sign.py b/backend/apps/tiktok_mcp_shim/tiktok_sign.py new file mode 100644 index 00000000..b23e85b9 --- /dev/null +++ b/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) diff --git a/backend/apps/tiktok_mcp_shim/tiktok_writes.py b/backend/apps/tiktok_mcp_shim/tiktok_writes.py new file mode 100644 index 00000000..09a89737 --- /dev/null +++ b/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()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()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)." + ), + } diff --git a/backend/apps/tiktok_mcp_shim/tools.py b/backend/apps/tiktok_mcp_shim/tools.py new file mode 100644 index 00000000..4ffe907b --- /dev/null +++ b/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/ 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"}, + }, + }, + }, +] diff --git a/backend/tests/test_tiktok_mcp_shim.py b/backend/tests/test_tiktok_mcp_shim.py new file mode 100644 index 00000000..1b42b07c --- /dev/null +++ b/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