LangChain

pip install langchain-mcp-adapters langgraph langchain-openai klavis

import os

# Set environment variables
os.environ["OPENAI_API_KEY"] = "your-openai-api-key-here"    # Replace with your actual OpenAI API key
os.environ["KLAVIS_API_KEY"] = "your-klavis-api-key-here"   # Replace with your actual Klavis API key

import os
import asyncio
from klavis import Klavis
from klavis.types import McpServerName, ConnectionType
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# Initialize clients
klavis_client = Klavis(api_key=os.getenv("KLAVIS_API_KEY"))
llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))

# Create a YouTube MCP server and get the server URL
youtube_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.YOUTUBE,
    user_id="1234",
    platform_name="klavis",
    connection_type=ConnectionType.STREAMABLE_HTTP,
)

youtube_mcp_server_url = youtube_mcp_instance.server_url
print(f"🔗 YouTube MCP server created at: {youtube_mcp_server_url}")

# Create MCP client with YouTube server
mcp_client = MultiServerMCPClient({
    "youtube": {
        "transport": "streamable_http",
        "url": youtube_mcp_server_url
    }
})

# Get tools from MCP server
tools = asyncio.run(mcp_client.get_tools())

# Create agent with MCP-based tools
youtube_agent = create_react_agent(
    model=llm,
    tools=tools,
    prompt="You are an AI assistant that uses MCP tools to analyze YouTube videos."
)

print("🤖 YouTube AI agent created successfully!")

YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v=LCEmiRjPEtQ"  # pick a video you like!

response = asyncio.run(youtube_agent.ainvoke({
    "messages": [{"role": "user", "content": f"Summarize this video: {YOUTUBE_VIDEO_URL}"}]
}))
print("✅ Video Summary:", response["messages"][-1].content)

import webbrowser

# Create YouTube MCP server
youtube_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.YOUTUBE,
    user_id="1234",
    platform_name="klavis",
    connection_type=ConnectionType.STREAMABLE_HTTP,
)

# Create Gmail MCP server with OAuth authorization
gmail_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.GMAIL,
    user_id="1234",
    platform_name="klavis",
    connection_type=ConnectionType.STREAMABLE_HTTP,
)

print("✅ Created YouTube and Gmail MCP instances")

# Open Gmail OAuth authorization
webbrowser.open(gmail_mcp_instance.oauth_url)
print(f"🔐 Opening OAuth authorization for Gmail, if you are not redirected, please open the following URL in your browser: {gmail_mcp_instance.oauth_url}")

from langgraph.graph import StateGraph, MessagesState
from typing import Annotated, Literal
from langchain_core.messages import BaseMessage

# Get MCP server URLs
youtube_mcp_server_url = youtube_mcp_instance.server_url
gmail_mcp_server_url = gmail_mcp_instance.server_url

# Create a single MCP client with both servers
mcp_client = MultiServerMCPClient({
    "youtube": {
        "transport": "streamable_http",
        "url": youtube_mcp_server_url
    },
    "gmail": {
        "transport": "streamable_http", 
        "url": gmail_mcp_server_url
    }
})

# Get tools from all MCP servers
all_tools = asyncio.run(mcp_client.get_tools())

# Create agents with access to all tools
youtube_agent = create_react_agent(
    model=llm,
    tools=all_tools,
    prompt="You are a YouTube video summarization expert. Use MCP tools to analyze and summarize videos. After summarizing, pass the summary to the gmail agent."
)

gmail_agent = create_react_agent(
    model=llm,
    tools=all_tools,
    prompt="You are an email assistant. Use MCP tools to send emails via Gmail."
)

print("🤖 Multi-agent workflow created with YouTube and Gmail agents!")

# Define workflow state
class WorkflowState(MessagesState):
    summary: str = ""
    email_sent: bool = False

# Define workflow functions
def youtube_summarizer(state: WorkflowState):
    """Summarize YouTube video"""
    response = asyncio.run(youtube_agent.ainvoke(state))
    return {
        "messages": response["messages"],
        "summary": response["messages"][-1].content
    }

def email_sender(state: WorkflowState):
    """Send email with summary"""
    email_prompt = f"Send the following summary via email: {state['summary']}"
    response = asyncio.run(gmail_agent.ainvoke({
        "messages": [{"role": "user", "content": email_prompt}]
    }))
    return {
        "messages": response["messages"],
        "email_sent": True
    }

# Create the workflow graph
workflow = StateGraph(WorkflowState)
workflow.add_node("youtube_summarizer", youtube_summarizer)
workflow.add_node("email_sender", email_sender)

# Define the flow
workflow.set_entry_point("youtube_summarizer")
workflow.add_edge("youtube_summarizer", "email_sender")
workflow.set_finish_point("email_sender")

# Compile the workflow
app = workflow.compile()

print("📊 Multi-agent workflow graph created!")

YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v=LCEmiRjPEtQ"  # pick a video you like!
EMAIL_RECIPIENT = "example@email.com"  # Replace with your email

initial_state = {
    "messages": [{
        "role": "user", 
        "content": f"Summarize this video {YOUTUBE_VIDEO_URL} and send it to {EMAIL_RECIPIENT}"
    }]
}

# Run the workflow
result = asyncio.run(app.ainvoke(initial_state))

print("\n✅ Workflow Result:", result)