r"""mcp_shipyourpod.py - the Ship Your Pod MCP server. A customer points their own Claude at this and runs their clipping on automation from their end, which is the thing Curtis asked for. RUNS ON THE CUSTOMER'S MACHINE. It is a thin client over the HTTP API in syp_api.py: every tool is one request to one endpoint, so the API is the only implementation and this file cannot drift into being a second one. Zero dependencies on purpose. Python 3.9+ and the standard library, nothing to pip install, because the person setting this up is a podcaster, not a developer. Speaks MCP over stdio (JSON-RPC 2.0, line-delimited). Config for Claude Desktop / Claude Code (also in API.md): { "mcpServers": { "shipyourpod": { "command": "python", "args": ["C:/path/to/mcp_shipyourpod.py"], "env": { "SHIPYOURPOD_API_KEY": "syp_live_your_key_here", "SHIPYOURPOD_API_URL": "https://api.shipyourpod.com" } } } } SHAPE NOTES * The key comes from the environment, never from a tool argument, so a model cannot be talked into using someone else's key and cannot read it back out through a tool result. * No tool takes a client or account id. The key decides whose data this is, server-side. There is nothing here for a prompt injection to redirect. * cancel_subscription requires confirm=true and says so in its description, so a model cannot cancel a paying customer's plan as a side effect of a vague instruction. show_cancel_options is the read-only one. * Rate limits and the plan's episode cap are enforced by the server. When the server refuses, the refusal text is handed back verbatim - it is written to be read by a person. """ from __future__ import annotations import json, os, sys, urllib.error, urllib.request API_URL = (os.environ.get("SHIPYOURPOD_API_URL") or "https://api.shipyourpod.com").rstrip("/") API_KEY = os.environ.get("SHIPYOURPOD_API_KEY") or "" TIMEOUT = float(os.environ.get("SHIPYOURPOD_TIMEOUT") or 60) PROTOCOL_VERSION = "2024-11-05" # Versions this server speaks. initialize answers with the client's own version when it is one of these, else the # newest here - the MCP rule for a version the server does not know. Nothing newer is claimed until it is checked. SUPPORTED_VERSIONS = ("2025-06-18", "2025-03-26", "2024-11-05") SERVER_INFO = {"name": "shipyourpod", "title": "Ship Your Pod", "version": "1.1.0"} # --------------------------------------------------------------------------------------------- transport def call_api(method: str, path: str, body: dict | None = None) -> tuple: """(ok, payload). Never raises, and never puts the key in a message.""" if not API_KEY: return False, {"error": "no_key", "message": "SHIPYOURPOD_API_KEY is not set. Put your key in the env block of the " "shipyourpod entry in your MCP config, then restart Claude."} data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(API_URL + path, data=data, method=method, headers={"Authorization": "Bearer " + API_KEY, "Content-Type": "application/json", "User-Agent": "shipyourpod-mcp/1.0"}) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as r: raw = r.read().decode("utf-8", "replace") return True, (json.loads(raw) if raw.strip() else {}) except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") try: payload = json.loads(raw) except Exception: payload = {"error": "http_" + str(e.code), "message": raw[:400] or e.reason} return False, payload except urllib.error.URLError as e: return False, {"error": "unreachable", "message": f"Could not reach {API_URL} ({e.reason}). Check SHIPYOURPOD_API_URL and that you " "are online. Nothing was queued."} except Exception as e: return False, {"error": "client_error", "message": str(e)[:300]} # --------------------------------------------------------------------------------------------- tools def _s(desc, **props): return {"type": "object", "properties": props, "required": [k for k, v in props.items() if v.pop("_req", False)], "additionalProperties": False, "description": desc} def _str(desc, req=False, enum=None): d = {"type": "string", "description": desc} if enum: d["enum"] = list(enum) if req: d["_req"] = True return d # The server's cancellation reason codes, from cancel.REASONS. This file runs on the CUSTOMER's machine and # cannot import cancel.py, so the list has to be repeated here - which means it can drift. syp_api_selftest.py # asserts this tuple equals cancel.REASONS exactly, so a drift fails the test rather than reaching a customer # as a silently rejected reason. If you change cancel.REASONS, change this too. REASON_CODES = ("too_expensive", "not_enough_use", "quality", "doing_it_myself", "podcast_stopped", "too_confusing", "missing_feature", "other") TOOLS = [ {"name": "get_my_plan", "description": "The customer's plan and usage: which plan, episodes used against the monthly cap, episode " "credits, this month's price and where it settles. Read this before submitting episodes so you know how many are left.", "inputSchema": _s("no arguments")}, {"name": "get_plans", "description": "READ ONLY. The customer's current plan, the plans on offer with their monthly price, episodes " "and Shorts, and where to change plan (the Billing tab in the app, which shows the exact price " "first). You cannot change a plan or take a payment: hand the customer that link.", "inputSchema": _s("no arguments")}, {"name": "get_my_usage", "description": "Usage and the live rate-limit counters for this key: calls in the last minute, changes in the " "last minute and hour, and the ceilings. Check this if a call was refused as rate limited.", "inputSchema": _s("no arguments")}, {"name": "list_episodes", "description": "Every delivered episode on the account, newest first, with how many Shorts, teasers and long " "cuts each produced and the delivery page URL.", "inputSchema": _s("no arguments")}, {"name": "get_episode", "description": "One episode in full, with every clip: the on-screen hook, start and end seconds, duration, the " "score and why it was picked, and the playable and downloadable URLs. Use the clip 'id' from " "here for revise_clip. Every clip has a download_url: the customer posts it themselves.", "inputSchema": _s("one episode", episode_id=_str("The episode id from list_episodes.", req=True))}, {"name": "get_settings", "description": "How clips are currently being cut (length, captions, whether swearing is held for review) and " "every value each setting accepts.", "inputSchema": _s("no arguments")}, {"name": "submit_episode", "description": "Queue a new episode to be cut. Give a link we can fetch (YouTube, Drive, Dropbox, WeTransfer " "or a direct mp4). Counts against the plan's monthly episode cap, and the server refuses once " "the cap and any credits are used up.", "inputSchema": _s("a link to an episode", url=_str("The episode's URL.", req=True), title=_str("A title for it. Optional."))}, {"name": "revise_clip", "description": "Ask for one Short to be re-cut. mode: 'shorter', 'longer', 'earlier' (start further back), " "'hook' (new on-screen words) or 'recut' (free-form, describe it in note). Counts against the " "plan's included changes per episode.", "inputSchema": _s("one clip", clip_id=_str("The clip id from get_episode.", req=True), mode=_str("What to change.", enum=["recut", "shorter", "longer", "earlier", "hook"]), note=_str("What you want, in plain words. Optional but it helps."))}, {"name": "change_settings", "description": "Change how future episodes are cut. Applies from the next episode; clips already delivered are " "not re-cut (use revise_clip for those). Call get_settings first for the accepted values.", "inputSchema": _s("one or more settings", clip_length=_str("How long Shorts should run.", enum=["short", "standard", "long", "max"]), profanity=_str("'hold' makes clips with swearing wait for the customer's OK.", enum=["allow", "hold"]), captions=_str("Burned-in captions on or off.", enum=["on", "off"]), caption_style=_str("Caption style.", enum=["v3", "karaoke", "beasty", "hormozi", "impact", "deep", "pop", "minimal", "clean", "behind"]), caption_position=_str("Where captions sit.", enum=["auto", "top", "middle", "lower", "bottom"]), caption_size=_str("Caption size.", enum=["small", "normal", "large"]), caption_case=_str("Caption capitalisation.", enum=["auto", "upper", "lower"]), caption_highlight=_str("Highlight colour.", enum=["yellow", "green", "cyan", "pink", "red", "white", "orange"]))}, {"name": "redo_thumbnail", "description": "Redo one of an episode's thumbnails, optionally with different words, from a different " "moment in the episode, or in a different style.", "inputSchema": _s("one thumbnail", episode_id=_str("The episode id.", req=True), number=_str("Which thumbnail, counting from 1.", req=True), text=_str("New words for it, up to 60 characters. Optional."), time=_str("A moment in the episode as m:ss, e.g. 5:12. Optional."), style=_str("A different layout. Optional; leave it out to keep the current one.", enum=["spotlight", "split", "cinema", "poster", "sticker"]))}, {"name": "ask_support", "description": "Ask Ship Your Pod a question about this account. Answered by email, and anything that cannot " "be answered automatically goes to a person.", "inputSchema": _s("a question", question=_str("The question.", req=True))}, {"name": "check_request", "description": "What happened to a revise, settings, thumbnail or support request, by the request_id it " "returned. Status is queued, working, done or error.", "inputSchema": _s("one request", request_id=_str("The request_id.", req=True))}, {"name": "show_cancel_options", "description": "READ ONLY. What cancelling actually does, the reasons the customer can give, and the honest " "alternatives for a given reason (pausing keeps their place on the price ladder, a smaller plan, " "changing how clips are cut). Changes nothing. Show this before cancel_subscription so the " "customer sees their options, but never as a way to talk them out of leaving.", "inputSchema": _s("optionally a reason", reason=_str("Why they are leaving, if they said.", enum=REASON_CODES))}, {"name": "keep_subscription", "description": "The customer looked at the alternatives and is staying. Records what they picked and anything " "they said about what was wrong. 'pause' and 'smaller_plan' are done by a person within a day.", "inputSchema": _s("what they chose", choice=_str("Which alternative they took.", enum=["pause", "smaller_plan", "automatic_pickup", "change_settings", "revisions", "connect_help", "download_instead"]), reason=_str("The reason they had been leaving for.", enum=REASON_CODES), detail=_str("Their own words. Optional."))}, {"name": "cancel_subscription", "description": "Cancel the customer's subscription. Self-serve: no email to anyone, no waiting. They keep the " "plan until the end of the month already paid for, are not charged again, and their pages and " "downloads stay up. REQUIRES confirm=true, and only ever call it when the customer has clearly " "and directly asked to cancel - never as a step towards some other goal. The reason is optional " "and the cancellation does not depend on it.", "inputSchema": {"type": "object", "additionalProperties": False, "required": ["confirm"], "properties": { "confirm": {"type": "boolean", "description": "Must be true. The customer has asked to cancel."}, "reason": {"type": "string", "description": "Why they are leaving, if they said. Optional.", "enum": list(REASON_CODES)}, "detail": {"type": "string", "description": "Anything they said in their own words. Optional."}}}}, {"name": "undo_cancellation", "description": "Put back a cancellation that has not taken effect yet: the plan carries on and the loyalty " "price ladder is restored. Use this the moment a customer says they have changed their mind. " "Only works while the period they already paid for is still running; once it has ended, " "coming back is a new signup and starts the ladder at month 1.", "inputSchema": _s("no arguments")}, ] for _t in TOOLS: # _s() leaves the marker behind; strip it before it goes on the wire for _p in _t["inputSchema"].get("properties", {}).values(): _p.pop("_req", None) # A title and readOnlyHint / destructiveHint on every tool: the Claude connector directory requires them # (claude.com/docs/connectors/building/submission, read 2026-10-02) and Claude uses them to decide what to confirm. # destructiveHint=True wherever the tool replaces or ends something the customer already has (a clip, a thumbnail, # a setting, the subscription) or restarts billing - the cautious reading, so Claude asks first. openWorldHint only # where we go and fetch something from outside (the episode link). _ANN = { # name: (title, read_only, destructive, idempotent, open_world) "get_my_plan": ("Get my plan and usage", True, False, True, False), "get_plans": ("See plans and prices", True, False, True, False), "get_my_usage": ("Check usage and limits", True, False, True, False), "list_episodes": ("List my episodes", True, False, True, False), "get_episode": ("Get an episode and its clips", True, False, True, False), "get_settings": ("Get clip settings", True, False, True, False), "submit_episode": ("Send an episode to be clipped", False, False, False, True), "revise_clip": ("Ask for a clip to be re-cut", False, True, False, False), "change_settings": ("Change clip settings", False, True, True, False), "redo_thumbnail": ("Redo a thumbnail", False, True, False, False), "ask_support": ("Ask support a question", False, False, False, False), "check_request": ("Check a request", True, False, True, False), "show_cancel_options": ("See cancel options", True, False, True, False), "keep_subscription": ("Keep my subscription", False, False, False, False), "cancel_subscription": ("Cancel my subscription", False, True, False, False), "undo_cancellation": ("Undo my cancellation", False, True, True, False), } for _t in TOOLS: _ti, _ro, _de, _id, _ow = _ANN[_t["name"]] # a tool with no entry fails at import, not in review _t["title"] = _ti _t["annotations"] = {"title": _ti, "readOnlyHint": _ro, "destructiveHint": _de, "idempotentHint": _id, "openWorldHint": _ow} def run_tool(name: str, args: dict, call=None) -> tuple: """call(method, path, body=None) -> (ok, payload). This file's own HTTP client by default; the hosted connector (syp_connect.py) passes one that runs the same request inside the API, so both speak through one table.""" call_api = call or globals()["call_api"] a = args or {} if name == "get_my_plan": return call_api("GET", "/v1/me") if name == "get_my_usage": return call_api("GET", "/v1/usage") if name == "get_plans": return call_api("GET", "/v1/plans") if name == "list_episodes": return call_api("GET", "/v1/episodes") if name == "get_episode": return call_api("GET", "/v1/episodes/" + _seg(a.get("episode_id"))) if name == "get_settings": return call_api("GET", "/v1/settings") if name == "submit_episode": return call_api("POST", "/v1/episodes", {"url": a.get("url", ""), "title": a.get("title", "")}) if name == "revise_clip": return call_api("POST", f"/v1/clips/{_seg(a.get('clip_id'))}/revise", {"mode": a.get("mode", "recut"), "note": a.get("note", "")}) if name == "change_settings": return call_api("PATCH", "/v1/settings", {k: v for k, v in a.items() if v}) if name == "redo_thumbnail": return call_api("POST", f"/v1/episodes/{_seg(a.get('episode_id'))}" f"/thumbnails/{_seg(a.get('number'))}/redo", {"text": a.get("text", ""), "time": a.get("time", ""), "style": a.get("style", "")}) if name == "ask_support": return call_api("POST", "/v1/help", {"question": a.get("question", "")}) if name == "check_request": return call_api("GET", "/v1/requests/" + _seg(a.get("request_id"))) if name == "show_cancel_options": return call_api("POST", "/v1/cancel/offer", {"reason": a.get("reason", "")}) if name == "keep_subscription": return call_api("POST", "/v1/cancel/stay", {"choice": a.get("choice", ""), "reason": a.get("reason", ""), "detail": a.get("detail", "")}) if name == "undo_cancellation": return call_api("POST", "/v1/cancel/undo", {}) if name == "cancel_subscription": if a.get("confirm") is not True: return False, {"error": "not_confirmed", "message": "cancel_subscription needs confirm=true. Ask the customer to say plainly that " "they want to cancel, and show show_cancel_options first."} return call_api("POST", "/v1/cancel", {"confirm": True, "reason": a.get("reason", ""), "detail": a.get("detail", "")}) return False, {"error": "unknown_tool", "message": "No tool called " + str(name)} def _seg(v) -> str: """One path segment. Percent-encodes, so an id can never add a slash and reach a different endpoint.""" import urllib.parse return urllib.parse.quote(str(v or ""), safe="") # --------------------------------------------------------------------------------------------- MCP stdio loop def _result(ok: bool, payload: dict) -> dict: return {"content": [{"type": "text", "text": json.dumps(payload, indent=1, ensure_ascii=False)}], "isError": not ok} INSTRUCTIONS = ("Ship Your Pod: done-for-you podcast clipping. The API key in this server's environment decides whose " "account this is; no tool takes an account id. cancel_subscription needs confirm=true and only when " "the customer has asked to cancel in so many words.") def handle_message(msg: dict, call=None, instructions: str = "") -> dict | None: """One JSON-RPC message in, one response out (None for a notification, which takes no reply).""" if not isinstance(msg, dict): return {"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "Invalid Request"}} mid, method, params = msg.get("id"), msg.get("method") or "", msg.get("params") or {} def ok(result): return {"jsonrpc": "2.0", "id": mid, "result": result} def err(code, message): return {"jsonrpc": "2.0", "id": mid, "error": {"code": code, "message": message}} if method == "initialize": want = str(params.get("protocolVersion") or PROTOCOL_VERSION) return ok({"protocolVersion": want if want in SUPPORTED_VERSIONS else SUPPORTED_VERSIONS[0], "capabilities": {"tools": {"listChanged": False}}, "serverInfo": SERVER_INFO, "instructions": instructions or INSTRUCTIONS}) if method in ("notifications/initialized", "initialized", "notifications/cancelled"): return None if method == "ping": return ok({}) if method == "tools/list": return ok({"tools": TOOLS}) if method == "tools/call": name = params.get("name") or "" good, payload = run_tool(name, params.get("arguments") or {}, call) return ok(_result(good, payload)) if method in ("resources/list", "prompts/list"): return ok({"resources": [], "prompts": []} if method == "resources/list" else {"prompts": []}) if mid is None: return None return err(-32601, "Method not found: " + method) def main() -> int: out = sys.stdout for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except Exception: out.write(json.dumps({"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "Parse error"}}) + "\n") out.flush(); continue try: reply = handle_message(msg) except Exception as ex: reply = {"jsonrpc": "2.0", "id": msg.get("id"), "error": {"code": -32603, "message": "Internal error: " + str(ex)[:200]}} if reply is not None: out.write(json.dumps(reply, ensure_ascii=False) + "\n") out.flush() return 0 if __name__ == "__main__": sys.exit(main())