Full Behavior Changer MCP agent

app.py

@@ -1,37 +1,337 @@
+import os
+import json
+import uuid
+import datetime
+from typing import Optional, Dict, Any
+
 import gradio as gr
+import yt_dlp
+from openai import OpenAI
+
+# =====================================
+# CONFIG
+# =====================================
+
+DATA_DIR = "/data/contracts"
+os.makedirs(DATA_DIR, exist_ok=True)
+
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+client = OpenAI(api_key=OPENAI_API_KEY)
+
+
+# =====================================
+# STORAGE UTILS
+# =====================================
+
+def _contract_path(uid: str) -> str:
+    return os.path.join(DATA_DIR, f"{uid}.json")
+
+
+def save_contract(contract: Dict[str, Any]) -> None:
+    uid = contract["uid"]
+    path = _contract_path(uid)
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(contract, f, ensure_ascii=False, indent=2)
 
 
-def create_contract(link: str, category: str = "general") -> dict:
+def load_contract(uid: str) -> Dict[str, Any]:
+    path = _contract_path(uid)
+    if not os.path.exists(path):
+        raise ValueError(f"Contract with uid={uid} not found.")
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+# =====================================
+# CORE TOOLS
+# =====================================
+
+def transcribe_reel(link: str) -> str:
     """
-    Create a simple habit contract from a reel link.
+    Download a reel/video from a link (e.g. Instagram) and transcribe audio using OpenAI Whisper.
+    Returns plain text transcript.
+    """
+    if not OPENAI_API_KEY:
+        raise ValueError("OPENAI_API_KEY not set in environment.")
+
+    # Download audio/video with yt-dlp
+    tmp_id = str(uuid.uuid4())
+    out_tmpl = f"/tmp/{tmp_id}.%(ext)s"
+
+    ydl_opts = {
+        "format": "mp4/bestaudio/best",
+        "outtmpl": out_tmpl,
+        "quiet": True,
+    }
+
+    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+        info = ydl.extract_info(link, download=True)
+        filename = ydl.prepare_filename(info)
+
+    # Transcription via OpenAI
+    with open(filename, "rb") as f:
+        transcript = client.audio.transcriptions.create(
+            model="gpt-4o-transcribe",
+            file=f
+        )
 
-    Args:
-        link: URL to the reel / content.
-        category: Area of life (e.g. career, health, relationships).
+    text = transcript.text if hasattr(transcript, "text") else str(transcript)
+    return text
 
-    Returns:
-        JSON with a fake uid, echo of params and first tiny step.
+
+def extract_habits(transcript: str, category: str = "general") -> Dict[str, Any]:
+    """
+    Analyze transcript and produce:
+    - goal (one sentence)
+    - 7 simple actionable steps (list of strings)
     """
-    if not link:
-        return {"error": "link is required"}
+    if not OPENAI_API_KEY:
+        raise ValueError("OPENAI_API_KEY not set in environment.")
+
+    if not transcript or not transcript.strip():
+        # Fallback generic plan
+        return {
+            "goal": f"Improve your {category} through one small action per day.",
+            "steps": [
+                "Pick one tiny action related to your goal and do it today.",
+                "Repeat the same action and make it slightly harder.",
+                "Share your progress with someone (message or note).",
+                "Add 10–15 minutes of focused deep work on this goal.",
+                "Remove one distraction that blocks this goal.",
+                "Create a simple checklist for next week.",
+                "Summarize what you learned in 3–5 sentences."
+            ]
+        }
+
+    system_msg = (
+        "You are a behavior change coach. "
+        "User gives you a transcript of a short motivational / educational video. "
+        "Your job is to extract ONE clear goal and EXACTLY 7 simple daily actions "
+        "that move the user toward that goal. "
+        "Return STRICT JSON with keys: goal (string), steps (array of 7 short strings). "
+        "Do NOT add any explanation, only JSON."
+    )
+
+    user_msg = (
+        f"Category: {category}\n\n"
+        f"Transcript (may be truncated):\n"
+        f"{transcript[:4000]}"
+    )
+
+    resp = client.chat.completions.create(
+        model="gpt-4.1-mini",
+        messages=[
+            {"role": "system", "content": system_msg},
+            {"role": "user", "content": user_msg},
+        ],
+        temperature=0.2,
+    )
+
+    raw = resp.choices[0].message.content
+
+    # Try to parse JSON even if model wrapped it in text/code fences
+    raw = raw.strip()
+    start = raw.find("{")
+    end = raw.rfind("}")
+    if start == -1 or end == -1:
+        raise ValueError(f"Cannot parse JSON from model output: {raw}")
+    json_str = raw[start:end + 1]
+
+    data = json.loads(json_str)
+
+    # Minimal validation / normalization
+    goal = data.get("goal") or f"Improve your {category}."
+    steps = data.get("steps") or []
+    steps = [s for s in steps if isinstance(s, str)]
+    if len(steps) < 7:
+        # pad if needed
+        while len(steps) < 7:
+            steps.append("Repeat a small action that moves you forward.")
+    else:
+        steps = steps[:7]
+
+    return {"goal": goal, "steps": steps}
+
+
+def create_contract(
+    link: str,
+    category: str = "general",
+    uid: Optional[str] = None,
+    auto_transcribe: bool = True
+) -> Dict[str, Any]:
+    """
+    Create a habit contract from a video link.
+
+    - Downloads + transcribes the video (if auto_transcribe=True)
+    - Extracts goal + 7 habit steps
+    - Saves contract to storage
+    - Returns public contract summary
+    """
+    if uid is None or not uid.strip():
+        uid = f"{category}_{uuid.uuid4().hex[:8]}"
+
+    transcript = ""
+    if auto_transcribe:
+        try:
+            transcript = transcribe_reel(link)
+        except Exception as e:
+            # Fallback: no transcript, but do not fail hard
+            transcript = ""
+            print(f"[WARN] Transcription failed: {e}")
+
+    habits = extract_habits(transcript, category=category)
+
+    contract = {
+        "uid": uid,
+        "link": link,
+        "category": category,
+        "goal": habits["goal"],
+        "steps": habits["steps"],
+        "current_index": 0,
+        "history": [],
+        "created_at": datetime.datetime.utcnow().isoformat() + "Z",
+    }
+
+    save_contract(contract)
 
     return {
-        "uid": f"demo_{category}",
+        "uid": uid,
         "link": link,
         "category": category,
-        "next_step": "Watch the reel and take ONE practical action today."
+        "goal": contract["goal"],
+        "first_step": contract["steps"][0],
+        "total_steps": len(contract["steps"]),
+    }
+
+
+def next_step(uid: str) -> Dict[str, Any]:
+    """
+    Return the next actionable step for given contract.
+    """
+    contract = load_contract(uid)
+    idx = contract.get("current_index", 0)
+    steps = contract.get("steps", [])
+
+    if idx >= len(steps):
+        return {
+            "uid": uid,
+            "done": True,
+            "message": "All steps completed.",
+            "goal": contract.get("goal", ""),
+        }
+
+    return {
+        "uid": uid,
+        "done": False,
+        "goal": contract.get("goal", ""),
+        "step_index": idx,
+        "step": steps[idx],
+        "remaining_steps": len(steps) - idx,
+    }
+
+
+def mark_done(uid: str) -> Dict[str, Any]:
+    """
+    Mark current step as done and advance to next one.
+    """
+    contract = load_contract(uid)
+    idx = contract.get("current_index", 0)
+    steps = contract.get("steps", [])
+
+    now = datetime.datetime.utcnow().isoformat() + "Z"
+
+    if idx < len(steps):
+        current_step = steps[idx]
+        contract.setdefault("history", []).append(
+            {"index": idx, "step": current_step, "done_at": now}
+        )
+        contract["current_index"] = idx + 1
+        save_contract(contract)
+
+    return next_step(uid)
+
+
+def report(uid: str) -> Dict[str, Any]:
+    """
+    Return a summary of contract progress.
+    """
+    contract = load_contract(uid)
+    steps = contract.get("steps", [])
+    idx = contract.get("current_index", 0)
+    history = contract.get("history", [])
+
+    total = len(steps)
+    done = min(idx, total)
+    remaining = max(total - done, 0)
+    progress = (done / total * 100.0) if total > 0 else 0.0
+
+    return {
+        "uid": uid,
+        "goal": contract.get("goal", ""),
+        "category": contract.get("category", ""),
+        "total_steps": total,
+        "done_steps": done,
+        "remaining_steps": remaining,
+        "progress_percent": round(progress, 1),
+        "history": history,
+        "current_step": steps[idx] if idx < total else None,
     }
 
 
+# =====================================
+# GRADIO UI + MCP EXPOSURE
+# =====================================
+
 with gr.Blocks() as demo:
-    gr.Markdown("# Behavior Changer – MCP Demo")
+    gr.Markdown(
+        "# Behavior Changer – Habit Agent (MCP)\n"
+        "Tworzy kontrakty nawyków z linków do Reels / video, "
+        "rozbija na kroki i udostępnia jako MCP tools."
+    )
+
+    with gr.Tab("Create Contract"):
+        link_in = gr.Textbox(label="Reel / Video URL")
+        category_in = gr.Textbox(label="Category", value="career")
+        uid_in = gr.Textbox(label="Custom UID (optional)", value="")
+        output_create = gr.JSON(label="Contract summary")
+
+        btn_create = gr.Button("Create Contract")
+        btn_create.click(
+            create_contract,
+            inputs=[link_in, category_in, uid_in],
+            outputs=output_create,
+        )
+
+    with gr.Tab("Next Step"):
+        uid_next_in = gr.Textbox(label="Contract UID")
+        output_next = gr.JSON(label="Next step")
+        btn_next = gr.Button("Get Next Step")
+        btn_next.click(next_step, inputs=[uid_next_in], outputs=output_next)
+
+    with gr.Tab("Mark Done"):
+        uid_done_in = gr.Textbox(label="Contract UID")
+        output_done = gr.JSON(label="After marking done (next step)")
+        btn_done = gr.Button("Mark Current Step as Done")
+        btn_done.click(mark_done, inputs=[uid_done_in], outputs=output_done)
 
-    link = gr.Textbox(label="Reel/link", placeholder="https://instagram.com/reel/...")
-    category = gr.Textbox(label="Category", value="career")
-    out = gr.JSON(label="Result JSON")
+    with gr.Tab("Report"):
+        uid_report_in = gr.Textbox(label="Contract UID")
+        output_report = gr.JSON(label="Progress report")
+        btn_report = gr.Button("Get Report")
+        btn_report.click(report, inputs=[uid_report_in], outputs=output_report)
 
-    btn = gr.Button("Generate Contract")
-    btn.click(create_contract, [link, category], out)
+    with gr.Tab("Debug: Transcribe Only"):
+        link_tr_in = gr.Textbox(label="Reel / Video URL")
+        output_tr = gr.Textbox(label="Transcript (preview)")
+        btn_tr = gr.Button("Transcribe")
+        btn_tr.click(transcribe_reel, inputs=[link_tr_in], outputs=output_tr)
 
+    # MCP tools – exposed via /gradio_api/mcp
+    gr.api(create_contract)
+    gr.api(next_step)
+    gr.api(mark_done)
+    gr.api(report)
+    gr.api(transcribe_reel)
+    gr.api(extract_habits)
 
 demo.launch(mcp_server=True)

requirements.txt

@@ -1 +1,3 @@
 gradio[mcp]
+yt-dlp
+openai>=1.40.0