Upload 4 files

app.py

@@ -0,0 +1,148 @@
+import streamlit as st
+from utils import run_agent_sync
+
+st.set_page_config(page_title="MCP POC", page_icon="🤖", layout="wide")
+
+st.title("Model Context Protocol(MCP) - Learning Path Generator")
+
+# Initialize session state for progress
+if 'current_step' not in st.session_state:
+    st.session_state.current_step = ""
+if 'progress' not in st.session_state:
+    st.session_state.progress = 0
+if 'last_section' not in st.session_state:
+    st.session_state.last_section = ""
+if 'is_generating' not in st.session_state:
+    st.session_state.is_generating = False
+
+# Sidebar for API and URL configuration
+st.sidebar.header("Configuration")
+
+# API Key input
+google_api_key = st.sidebar.text_input("Google API Key", type="password")
+
+# Pipedream URLs
+st.sidebar.subheader("Pipedream URLs")
+youtube_pipedream_url = st.sidebar.text_input("YouTube URL (Required)", 
+    placeholder="Enter your Pipedream YouTube URL")
+
+# Secondary tool selection
+secondary_tool = st.sidebar.radio(
+    "Select Secondary Tool:",
+    ["Drive", "Notion"]
+)
+
+# Secondary tool URL input
+if secondary_tool == "Drive":
+    drive_pipedream_url = st.sidebar.text_input("Drive URL", 
+        placeholder="Enter your Pipedream Drive URL")
+    notion_pipedream_url = None
+else:
+    notion_pipedream_url = st.sidebar.text_input("Notion URL", 
+        placeholder="Enter your Pipedream Notion URL")
+    drive_pipedream_url = None
+
+# Quick guide before goal input
+st.info("""
+**Quick Guide:**
+1. Enter your Google API key and YouTube URL (required)
+2. Select and configure your secondary tool (Drive or Notion)
+3. Enter a clear learning goal, for example:
+    - "I want to learn python basics in 3 days"
+    - "I want to learn data science basics in 10 days"
+""")
+
+# Main content area
+st.header("Enter Your Goal")
+user_goal = st.text_input("Enter your learning goal:",
+                        help="Describe what you want to learn, and we'll generate a structured path using YouTube content and your selected tool.")
+
+# Progress area
+progress_container = st.container()
+progress_bar = st.empty()
+
+def update_progress(message: str):
+    """Update progress in the Streamlit UI"""
+    st.session_state.current_step = message
+    
+    # Determine section and update progress
+    if "Setting up agent with tools" in message:
+        section = "Setup"
+        st.session_state.progress = 0.1
+    elif "Added Google Drive integration" in message or "Added Notion integration" in message:
+        section = "Integration"
+        st.session_state.progress = 0.2
+    elif "Creating AI agent" in message:
+        section = "Setup"
+        st.session_state.progress = 0.3
+    elif "Generating your learning path" in message:
+        section = "Generation"
+        st.session_state.progress = 0.5
+    elif "Learning path generation complete" in message:
+        section = "Complete"
+        st.session_state.progress = 1.0
+        st.session_state.is_generating = False
+    else:
+        section = st.session_state.last_section or "Progress"
+    
+    st.session_state.last_section = section
+    
+    # Show progress bar
+    progress_bar.progress(st.session_state.progress)
+    
+    # Update progress container with current status
+    with progress_container:
+        # Show section header if it changed
+        if section != st.session_state.last_section and section != "Complete":
+            st.write(f"**{section}**")
+        
+        # Show message with tick for completed steps
+        if message == "Learning path generation complete!":
+            st.success("All steps completed! 🎉")
+        else:
+            prefix = "✓" if st.session_state.progress >= 0.5 else "→"
+            st.write(f"{prefix} {message}")
+
+# Generate Learning Path button
+if st.button("Generate Learning Path", type="primary", disabled=st.session_state.is_generating):
+    if not google_api_key:
+        st.error("Please enter your Google API key in the sidebar.")
+    elif not youtube_pipedream_url:
+        st.error("YouTube URL is required. Please enter your Pipedream YouTube URL in the sidebar.")
+    elif (secondary_tool == "Drive" and not drive_pipedream_url) or (secondary_tool == "Notion" and not notion_pipedream_url):
+        st.error(f"Please enter your Pipedream {secondary_tool} URL in the sidebar.")
+    elif not user_goal:
+        st.warning("Please enter your learning goal.")
+    else:
+        try:
+            # Set generating flag
+            st.session_state.is_generating = True
+            
+            # Reset progress
+            st.session_state.current_step = ""
+            st.session_state.progress = 0
+            st.session_state.last_section = ""
+            
+            result = run_agent_sync(
+                google_api_key=google_api_key,
+                youtube_pipedream_url=youtube_pipedream_url,
+                drive_pipedream_url=drive_pipedream_url,
+                notion_pipedream_url=notion_pipedream_url,
+                user_goal=user_goal,
+                progress_callback=update_progress
+            )
+            
+            # Display results
+            st.header("Your Learning Path")
+            # print(result)
+            if result and "messages" in result:
+                for msg in result["messages"]:
+                    st.markdown(f"📚 {msg.content}")
+
+            else:
+                st.error("No results were generated. Please try again.")
+                st.session_state.is_generating = False
+        except Exception as e:
+            st.error(f"An error occurred: {str(e)}")
+            st.error("Please check your API keys and URLs, and try again.")
+            st.session_state.is_generating = False

prompt.py

@@ -0,0 +1,35 @@
+user_goal_prompt = """
+Main Instruction: You are a day wise learning path generator. You will be given a goal. You have to generate a comprehensive day-wise learning path for the user goal in a Drive document/Notion page and create a corresponding YouTube playlist containing the core learning videos.
+Step-by-Step Execution Flow:
+You must follow these steps sequentially to fulfill the user's request:
+1. Plan the Learning Path Structure: Internally, devise a day-wise structure of topics relevant to the user's learning goal. Determine the core topics needed to achieve the goal, aiming for a logical progression. Limit the number of days/topics to a manageable size for a foundational path.
+2. Research Potential Video Resources: For each topic identified in the plan, search and identify multiple relevant YouTube video URLs that could serve as learning resources. This step aims for broad discovery.
+3. Select Core Videos for the Learning Path: From the researched videos, select the single most suitable video for each day/topic in your planned structure (from Step 1). These selected videos should be foundational, provide excellent overviews, or be highly impactful for that specific topic and the overall learning goal. The total number of selected videos will match the number of days/topics planned. These selected videos will form the content of both the document and the playlist.
+4. Format the Document Content: Create the content that will go into the document/Notion page. Using the structure from Step 1 and the core videos selected in Step 3, format the day-wise learning path according to the 'Learning path sample format'. Ensure the content includes a clear main title for the learning path and uses headers/titles for each day/section.
+5. Create and Populate Drive Document/Notion Page:
+    a. Create a new document in Google Drive/Notion page(while creating get the document/notion page id for to read and write). choosing based on the tools available to you.
+    b. Paste the formatted day-wise learning path content from Step 4 into the document/Notion page. Ensure all YouTube links are properly formatted as clickable links within the document.
+    c. Save the document/Notion page ID for further use.
+6. Create Public YouTube Playlist:
+    a. Create one public YouTube playlist with a relevant title for the overall learning path.
+    b. Save the YouTube playlist ID for further use(to add youtube video urls from the learning path).
+    c. Add only the core videos selected in Step 3 (the same videos listed in the document) to this playlist.
+    if you are unable to find a previously created playlist by you, try the step 6 again.
+7. (Optional) Suggest Further Resources: (If deemed relevant for the topic based on your knowledge) Add a small section at the end of the document/Notion page suggesting "Top Channels or Institutes to Follow" for further learning on the main topic.
+8. Provide Outputs: Ensure the final response to the user includes the links to the created Google Drive document/Notion page and the YouTube playlist. The final output should explicitly state: "Here is your learning path document link: [link]" and "Here is your YouTube playlist link: [link] (with relevant content)".
+
+General Instructions & Guidelines:
+1. Act like a team player, coordinating between tools. 
+2. Utilize the provided tool descriptions. Choose tools like Google Drive/Notion and YouTube API based on their availability and your capabilities.
+3. You can use multiple tools simultaneously.
+4. Do not ask for confirmation from the user; proceed with the best possible outcome.
+5. If encountering errors (e.g., unable to edit), find alternatives (e.g., create a new document).
+6. When searching for resources, use terms users would generally use.
+7. Remember to track document/page and playlist IDs for potential future interactions.
+
+Learning path sample format within a day/section (to be used with overall document titles and headers):
+Day X:
+Topic: Topic name X
+YouTube Link: URL of the core video selected for Topic X
+(Continue for subsequent days...)
+"""

requirements.txt

@@ -1,3 +1,5 @@
-altair
-pandas
+langchain
+langgraph
+langchain-mcp-adapters
+langchain-google-genai
 streamlit

utils.py

@@ -0,0 +1,129 @@
+from langchain_core.messages import HumanMessage
+from langchain_core.runnables import RunnableConfig
+from prompt import user_goal_prompt
+from langgraph.prebuilt import create_react_agent
+from langchain_mcp_adapters.client import MultiServerMCPClient
+from langchain_google_genai import ChatGoogleGenerativeAI
+from typing import Optional, Tuple, Any, Callable
+import asyncio
+
+cfg = RunnableConfig(recursion_limit=100)
+
+def initialize_model(google_api_key: str) -> ChatGoogleGenerativeAI:
+    return ChatGoogleGenerativeAI(
+        model="gemini-2.5-flash",
+        google_api_key=google_api_key
+    )
+
+async def setup_agent_with_tools(
+    google_api_key: str,
+    youtube_pipedream_url: str,
+    drive_pipedream_url: Optional[str] = None,
+    notion_pipedream_url: Optional[str] = None,
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> Any:
+    """
+    Set up the agent with YouTube (mandatory) and optional Drive or Notion tools.
+    """
+    try:
+        if progress_callback:
+            progress_callback("Setting up agent with tools... ✅")
+        
+        # Initialize tools configuration with mandatory YouTube
+        tools_config = {
+            "youtube": {
+                "url": youtube_pipedream_url,
+                "transport": "streamable_http"
+            }
+        }
+
+        # Add Drive if URL provided
+        if drive_pipedream_url:
+            tools_config["drive"] = {
+                "url": drive_pipedream_url,
+                "transport": "streamable_http"
+            }
+            if progress_callback:
+                progress_callback("Added Google Drive integration... ✅")
+
+        # Add Notion if URL provided
+        if notion_pipedream_url:
+            tools_config["notion"] = {
+                "url": notion_pipedream_url,
+                "transport": "streamable_http"
+            }
+            if progress_callback:
+                progress_callback("Added Notion integration... ✅")
+
+        if progress_callback:
+            progress_callback("Initializing MCP client... ✅")
+        # Initialize MCP client with configured tools
+        mcp_client = MultiServerMCPClient(tools_config)
+        
+        if progress_callback:
+            progress_callback("Getting available tools... ✅")
+        # Get all tools
+        tools = await mcp_client.get_tools()
+        
+        if progress_callback:
+            progress_callback("Creating AI agent... ✅")
+        # Create agent with initialized model
+        mcp_orch_model = initialize_model(google_api_key)
+        agent = create_react_agent(mcp_orch_model, tools)
+        
+        if progress_callback:
+            progress_callback("Setup complete! Starting to generate learning path... ✅")
+        
+        return agent
+    except Exception as e:
+        print(f"Error in setup_agent_with_tools: {str(e)}")
+        raise
+
+def run_agent_sync(
+    google_api_key: str,
+    youtube_pipedream_url: str,
+    drive_pipedream_url: Optional[str] = None,
+    notion_pipedream_url: Optional[str] = None,
+    user_goal: str = "",
+    progress_callback: Optional[Callable[[str], None]] = None
+) -> dict:
+    """
+    Synchronous wrapper for running the agent.
+    """
+    async def _run():
+        try:
+            agent = await setup_agent_with_tools(
+                google_api_key=google_api_key,
+                youtube_pipedream_url=youtube_pipedream_url,
+                drive_pipedream_url=drive_pipedream_url,
+                notion_pipedream_url=notion_pipedream_url,
+                progress_callback=progress_callback
+            )
+            
+            # Combine user goal with prompt template
+            learning_path_prompt = "User Goal: " + user_goal + "\n" + user_goal_prompt
+            
+            if progress_callback:
+                progress_callback("Generating your learning path...")
+            
+            # Run the agent
+            result = await agent.ainvoke(
+                {"messages": [HumanMessage(content=learning_path_prompt)]},
+                config=cfg
+            )
+            
+            if progress_callback:
+                progress_callback("Learning path generation complete!")
+            
+            return result
+        except Exception as e:
+            print(f"Error in _run: {str(e)}")
+            raise
+
+    # Run in new event loop
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        return loop.run_until_complete(_run())
+    finally:
+        loop.close()