#!/usr/bin/env python3 """ Piped SDK - Simple Python client for executing pipelines (stdlib only, no pip install) @param pipeline: The pipeline string to execute (e.g., "cat | grep hello | wc -l") @param input: Optional stdin for the pipeline (default: ""). Pass bytes for binary input. @param key: API key. Falls back to PIPED_API_KEY env var if omitted. @param server_url: Optional server URL (default: "https://piped.sh") @return: The pipeline output as a string @raises: Exception If the request fails or returns an error Examples: piped("cat | grep hello") # key from PIPED_API_KEY env piped("cat | grep hello", "hello world\\nfoo bar") # with stdin piped("cat | grep hello", key="pk_...") # explicit key piped("cat | grep hello", "hello world\\nfoo bar", key="pk_...") CLI (run stand-alone: python3 piped.py [--mcp] ): --mcp Write .mcp.json in current directory (only if missing). Uses PIPED_API_KEY. No args + TTY: REPL mode (exit, quit). """ import json import os import sys import urllib.error import urllib.parse import urllib.request from datetime import datetime def piped( pipeline: str, input="", *, key: str = None, server_url: str = "https://piped.sh", ) -> str: if not pipeline or not isinstance(pipeline, str): raise ValueError("pipeline is required and must be a string") api_key = key or os.environ.get("PIPED_API_KEY", "") if not api_key: raise ValueError("API key is required: pass key='pk_...' or set PIPED_API_KEY env var") base = server_url.rstrip("/") headers = { "X-API-Key": api_key, "X-User-Time": datetime.now().astimezone().isoformat(), } if isinstance(input, bytes): # Raw body with pipeline in query; server detects type from magic bytes url = f"{base}/?{urllib.parse.urlencode({'pipeline': pipeline})}" req = urllib.request.Request(url, data=input, headers=headers, method="POST") req.add_header("Content-Type", "application/octet-stream") else: # JSON with pipeline + input (text) url = f"{base}/" body = json.dumps({"pipeline": pipeline, "input": input or ""}).encode("utf-8") req = urllib.request.Request(url, data=body, headers=headers, method="POST") req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req) as resp: return resp.read().decode("utf-8") except urllib.error.HTTPError as e: err_msg = f"Request failed with status {e.code}" try: err_data = json.loads(e.read().decode("utf-8")) err_msg = err_data.get("error", err_msg) except (json.JSONDecodeError, ValueError): err_msg = e.reason or err_msg raise Exception(err_msg) from e # Command-line interface - detect if script is run directly if __name__ == "__main__": server_url = os.environ.get("PIPED_SERVER_URL", "https://piped.sh").rstrip("/") api_key = os.environ.get("PIPED_API_KEY", "") repl_prompt = os.environ.get("PIPED_REPL_PROMPT", "piped> ") # --mcp: write .mcp.json in current directory (only if missing) if len(sys.argv) >= 2 and sys.argv[1] == "--mcp": mcp_path = os.path.join(os.getcwd(), ".mcp.json") if os.path.isfile(mcp_path): print(".mcp.json already exists", file=sys.stderr) sys.exit(0) api_key_for_mcp = api_key or "pk_your_api_key_here" mcp_json = { "mcpServers": { "piped": { "type": "http", "url": f"{server_url}/mcp", "headers": {"X-API-Key": api_key_for_mcp}, }, }, } with open(mcp_path, "w") as f: json.dump(mcp_json, f, indent=2) f.write("\n") print("Wrote .mcp.json", file=sys.stderr) sys.exit(0) no_args = len(sys.argv) == 1 if no_args and not sys.stdin.isatty(): print("Error: pipeline is required", file=sys.stderr) print(f"Usage: {sys.argv[0]} [--mcp] ", file=sys.stderr) print(" --mcp Write .mcp.json in current directory (only if missing).", file=sys.stderr) print("With no arguments and a TTY, enters REPL mode.", file=sys.stderr) if not api_key: print("Note: Set PIPED_API_KEY environment variable with your API key", file=sys.stderr) sys.exit(1) if no_args and sys.stdin.isatty(): # REPL mode (with readline-style up/down history when available) _has_readline = False try: import readline readline.set_history_length(100) _has_readline = True except ImportError: pass def read_line(): # When readline is imported, input() automatically supports history & arrow keys if _has_readline: return input(repl_prompt) sys.stdout.write(repl_prompt) sys.stdout.flush() raw = sys.stdin.readline() if not raw: raise EOFError return raw.rstrip("\n") while True: try: line = read_line() except (KeyboardInterrupt, EOFError): break l = line.strip().lstrip("/").strip() lower = l.lower() if lower in ("exit", "quit"): break if not l: continue if not api_key: print("Error: PIPED_API_KEY environment variable is required", file=sys.stderr) continue try: result = piped(l, key=api_key, server_url=server_url) print(result, end="" if result.endswith("\n") else "\n") if _has_readline and l: # Add to history; avoid duplicate of most recent try: n = readline.get_current_history_length() if n == 0 or readline.get_history_item(n) != l: readline.add_history(l) except (AttributeError, OSError): readline.add_history(l) except Exception as e: print(f"Error: {e}", file=sys.stderr) sys.exit(0) # Single-shot: pipeline provided as arguments # When invoked from piped.sh with one arg, that arg is the pre-joined pipeline (quotes preserved). # Otherwise quote each arg and join. argv_args = sys.argv[1:] # Handle "project --manage": POST /project?manage, then open browser if argv_args == ["project", "--config"]: if not api_key: print("Error: PIPED_API_KEY environment variable is required", file=sys.stderr) sys.exit(1) try: req = urllib.request.Request( f"{server_url}/project?manage", data=b"", headers={"X-API-Key": api_key}, method="POST", ) with urllib.request.urlopen(req) as resp: config_url = resp.read().decode("utf-8").strip() import webbrowser webbrowser.open(config_url) print(f"Opening project manager in browser: {config_url}") sys.exit(0) except Exception as e: print(f"Error: {e}", file=sys.stderr) sys.exit(1) if len(argv_args) == 1: pipeline = argv_args[0] else: def quote_arg(arg): if any(c in arg for c in ' "\'\\$`(){}[];&|<>'): escaped = arg.replace("\\", "\\\\").replace('"', '\\"') return f'"{escaped}"' return arg pipeline = " ".join(quote_arg(arg) for arg in argv_args) if not api_key: print("Error: PIPED_API_KEY environment variable is required", file=sys.stderr) print('Set it with: export PIPED_API_KEY="pk_your_api_key_here"', file=sys.stderr) sys.exit(1) input_data = "" if not sys.stdin.isatty(): input_data = sys.stdin.buffer.read() try: result = piped(pipeline, input_data, key=api_key, server_url=server_url) print(result) sys.exit(0) except Exception as e: print(f"Error: {e}", file=sys.stderr) sys.exit(1)