#!/bin/bash # # fetch-swagger.sh # Fetch an OpenAPI / Swagger spec from a URL, return a normalized JSON view # of the API surface that Phase 1 (Analysis) can prepend to the agent prompt. # # The output is intentionally small (≤ ~64KB target) - full schemas are # truncated to one example each, descriptions are trimmed to one paragraph, # and only the endpoints' shape is kept. Phase 1 doesn't need the entire # spec, it needs to know which endpoints exist, their method/path, and # the rough shape of request/response. # # Usage: # ./fetch-swagger.sh # # Optional env: # SWAGGER_TIMEOUT_SECONDS default 15 # SWAGGER_MAX_BYTES default 5_000_000 (5MB) - refuses payloads larger # SWAGGER_AUTH_HEADER optional 'Header-Name: value' line, when the spec # URL needs auth (rare; most OpenAPI specs are # served unauthenticated) # # Output (stdout, single JSON object): # { # "fetchedAt": "", # "source": { "url": "", "format": "json|yaml", "bytes": }, # "info": { "title": "...", "version": "...", "description": "..." }, # "servers": ["", ...], # "endpoints": [ # { "method": "POST", "path": "/v1/...", "summary": "...", # "requestExample": { ... }, # "responses": { "200": { "schemaSummary": "...", "example": {...} }, # "400": { "schemaSummary": "..." } } }, # ... # ], # "endpointCount": # } # # Exit codes: # 0 success # 2 network / HTTP error # 3 payload too large or malformed # 4 bad usage set -euo pipefail URL="${1:-}" [ -z "$URL" ] && { echo "usage: $0 " >&2; exit 4; } TIMEOUT="${SWAGGER_TIMEOUT_SECONDS:-15}" MAX_BYTES="${SWAGGER_MAX_BYTES:-5000000}" TMP_RAW=$(mktemp -t swagger-raw.XXXXXX) trap 'rm -f "$TMP_RAW"' EXIT # Pull the spec with a size cap (curl --max-filesize doesn't exist on every # platform; we let curl run with --output then check the size). CURL_ARGS=(-sS --fail --max-time "$TIMEOUT" --connect-timeout 5 --location --output "$TMP_RAW" \ --write-out "%{http_code}\n%{content_type}\n%{size_download}\n") # The optional auth header may carry a token, so it is fed through a curl # config via process substitution instead of argv (argv is visible to ps). swagger_auth_cfg() { printf 'header = "%s"\n' "$SWAGGER_AUTH_HEADER"; } if [ -n "${SWAGGER_AUTH_HEADER:-}" ]; then META=$(curl "${CURL_ARGS[@]}" -K <(swagger_auth_cfg) "$URL" 2>/dev/null) || { echo "ERR: swagger fetch failed for $URL" >&2 exit 2 } else META=$(curl "${CURL_ARGS[@]}" "$URL" 2>/dev/null) || { echo "ERR: swagger fetch failed for $URL" >&2 exit 2 } fi HTTP_CODE=$(printf '%s\n' "$META" | sed -n '1p') CONTENT_TYPE=$(printf '%s\n' "$META" | sed -n '2p') SIZE=$(printf '%s\n' "$META" | sed -n '3p') if [ "$HTTP_CODE" != "200" ]; then echo "ERR: swagger HTTP $HTTP_CODE for $URL" >&2 exit 2 fi if [ "${SIZE:-0}" -gt "$MAX_BYTES" ]; then echo "ERR: swagger payload $SIZE B exceeds max $MAX_BYTES B" >&2 exit 3 fi # Detect format from content-type or URL suffix; YAML→JSON conversion uses # python's yaml when available, falls back to "format=yaml" passthrough so # the agent can still read it if pyyaml is missing. FORMAT="json" LOWER_CT=$(printf '%s' "$CONTENT_TYPE" | tr '[:upper:]' '[:lower:]') case "$LOWER_CT" in *yaml*|*yml*) FORMAT="yaml" ;; *json*) FORMAT="json" ;; *) # Lowercase via tr: the bash-4 case-conversion expansion is not # available on macOS /bin/bash 3.2. LOWER_URL=$(printf '%s' "$URL" | tr '[:upper:]' '[:lower:]') case "$LOWER_URL" in *.yaml|*.yml) FORMAT="yaml" ;; *) FORMAT="json" ;; esac ;; esac SOURCE_URL="$URL" SOURCE_FORMAT="$FORMAT" SOURCE_BYTES="$SIZE" \ SOURCE_PATH="$TMP_RAW" \ python3 - <<'PY' import json, os, re, sys, datetime src_path = os.environ["SOURCE_PATH"] fmt = os.environ["SOURCE_FORMAT"] url = os.environ["SOURCE_URL"] size = int(os.environ.get("SOURCE_BYTES") or 0) def load_spec(): with open(src_path, "rb") as f: raw = f.read() text = raw.decode("utf-8", errors="replace") if fmt == "yaml": try: import yaml # type: ignore return yaml.safe_load(text) except ImportError: # Fall back to a minimal stub so the pipeline can still surface # the URL with a "yaml-no-parser" advisory. return {"info": {"title": ""}, "_unparsed": True} try: return json.loads(text) except json.JSONDecodeError as e: print(f"ERR: spec is not valid JSON: {e}", file=sys.stderr) sys.exit(3) spec = load_spec() if not isinstance(spec, dict): print("ERR: spec root is not an object", file=sys.stderr) sys.exit(3) def trim(text, n=240): if not text: return "" text = re.sub(r"\s+", " ", str(text)).strip() return text[:n] + ("..." if len(text) > n else "") def example_for(schema, components, depth=0): """Build a tiny example value from a JSON Schema fragment. Stops at depth 4 to avoid pathological circular references.""" if depth > 4 or not isinstance(schema, dict): return None if "$ref" in schema and isinstance(schema["$ref"], str): ref = schema["$ref"] # Local refs: #/components/schemas/Foo if ref.startswith("#/components/schemas/"): name = ref.rsplit("/", 1)[-1] target = (components or {}).get("schemas", {}).get(name) if isinstance(target, dict): return example_for(target, components, depth + 1) return f"<{ref}>" if "example" in schema: return schema["example"] if "examples" in schema and isinstance(schema["examples"], dict): first = next(iter(schema["examples"].values()), None) if isinstance(first, dict) and "value" in first: return first["value"] t = schema.get("type") if t == "object" or "properties" in schema: out = {} for name, sub in (schema.get("properties") or {}).items(): out[name] = example_for(sub, components, depth + 1) return out if t == "array": return [example_for(schema.get("items") or {}, components, depth + 1)] if t == "string": return schema.get("default", "") if t in ("number", "integer"): return schema.get("default", 0) if t == "boolean": return schema.get("default", False) return None def schema_summary(schema, components, depth=0): if depth > 3 or not isinstance(schema, dict): return "" if "$ref" in schema: return schema["$ref"].rsplit("/", 1)[-1] t = schema.get("type") if t == "object" or "properties" in schema: keys = list((schema.get("properties") or {}).keys())[:8] return "object{" + ", ".join(keys) + ("..." if len(keys) >= 8 else "") + "}" if t == "array": return "array<" + schema_summary(schema.get("items") or {}, components, depth + 1) + ">" return t or "" components = spec.get("components") or {} info_raw = spec.get("info") or {} info = { "title": trim(info_raw.get("title") or "", 120), "version": trim(info_raw.get("version") or "", 40), "description": trim(info_raw.get("description") or "", 400), } servers = [] for s in (spec.get("servers") or []): if isinstance(s, dict) and isinstance(s.get("url"), str): servers.append(s["url"]) # Swagger 2.0 fallback if not servers and spec.get("host"): scheme = (spec.get("schemes") or ["https"])[0] base = spec.get("basePath") or "" servers.append(f"{scheme}://{spec['host']}{base}") endpoints = [] paths = spec.get("paths") or {} for path, methods in paths.items(): if not isinstance(methods, dict): continue for method, op in methods.items(): if method.lower() not in {"get", "post", "put", "patch", "delete", "options", "head"}: continue if not isinstance(op, dict): continue # Request example req_example = None # OpenAPI 3.x rb = op.get("requestBody") if isinstance(rb, dict): content = (rb.get("content") or {}) primary = content.get("application/json") or next(iter(content.values()), {}) if isinstance(primary, dict) and primary.get("schema"): req_example = example_for(primary["schema"], components) # Swagger 2.0 elif isinstance(op.get("parameters"), list): for p in op["parameters"]: if isinstance(p, dict) and p.get("in") == "body" and p.get("schema"): req_example = example_for(p["schema"], components) break # Responses responses_out = {} for code, resp in (op.get("responses") or {}).items(): if not isinstance(resp, dict): continue schema = None # OpenAPI 3.x content = resp.get("content") or {} primary = content.get("application/json") or next(iter(content.values()), None) if isinstance(primary, dict) and primary.get("schema"): schema = primary["schema"] # Swagger 2.0 elif resp.get("schema"): schema = resp["schema"] responses_out[str(code)] = { "schemaSummary": schema_summary(schema, components) if schema else "", "example": example_for(schema, components) if schema else None, } # Cap at the 3 most relevant codes per endpoint if len(responses_out) >= 3: break endpoints.append({ "method": method.upper(), "path": path, "summary": trim(op.get("summary") or op.get("description") or "", 200), "requestExample": req_example, "responses": responses_out, }) result = { "fetchedAt": datetime.datetime.utcnow().isoformat() + "Z", "source": {"url": url, "format": fmt, "bytes": size}, "info": info, "servers": servers, "endpoints": endpoints, "endpointCount": len(endpoints), } print(json.dumps(result, ensure_ascii=False)) PY