Fireworks AI

pip install fireworks-ai klavis

import os

# Set environment variables
os.environ["FIREWORKS_API_KEY"] = "your-fireworks-api-key-here"  # Replace with your actual Fireworks API key
os.environ["KLAVIS_API_KEY"] = "your-klavis-api-key-here"       # Replace with your actual Klavis API key

import os
import json
from fireworks.client import Fireworks
from klavis import Klavis
from klavis.types import McpServerName, ToolFormat

# Initialize clients
fireworks_client = Fireworks(api_key=os.getenv("FIREWORKS_API_KEY"))
klavis_client = Klavis(api_key=os.getenv("KLAVIS_API_KEY"))

class Agent:
    def __init__(self, fireworks_client, klavis_client, mcp_server_url):
        self.fireworks = fireworks_client
        self.klavis = klavis_client
        self.mcp_server_url = mcp_server_url
        self.model = "accounts/fireworks/models/qwen2p5-72b-instruct"
        print(f"🤖 Agent initialized with model: {self.model}")
    
    def process_request(self, user_message):
        # 1. Get available tools
        mcp_tools = self.klavis.mcp_server.list_tools(
            server_url=self.mcp_server_url,
            format=ToolFormat.OPENAI
        )
        
        # 2. Call LLM with tools
        messages = [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": user_message}
        ]
        
        response = self.fireworks.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=mcp_tools.tools
        )
        
        assistant_message = response.choices[0].message
        messages.append(assistant_message)
        
        # 3. If LLM wants to use tools
        if assistant_message.tool_calls:
            
            # Execute each tool call
            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)
                
                print(f"🛠️ Calling tool: {tool_name} with args: {tool_args}")
                # Call tool via Klavis SDK
                tool_result = self.klavis.mcp_server.call_tools(
                    server_url=self.mcp_server_url,
                    tool_name=tool_name,
                    tool_args=tool_args
                )
                
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(tool_result)
                })
            
            # 4. Get final response from LLM
            final_response = self.fireworks.chat.completions.create(
                model=self.model,
                messages=messages
            )
            return final_response.choices[0].message.content
        
        # If no tools needed, return the assistant message directly
        return assistant_message.content

YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v=kPXvf2-C_Hs"  # Pick a video you like!

# 1. Create YouTube MCP server instance
youtube_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.YOUTUBE,
    user_id="1234"
)

# 2. Create an agent with YouTube MCP server
agent = Agent(fireworks_client, klavis_client, youtube_mcp_instance.server_url)

# 3. Process the request
response = agent.process_request(
    f"Summarize this YouTube video with timestamps: {YOUTUBE_VIDEO_URL}"
)

print(response)

import webbrowser

# Create Gmail MCP server instance
gmail_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.GMAIL,
    user_id="1234"
)

# Redirect to Gmail OAuth page
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}")

EMAIL_SUBJECT = "Hello, World!"
EMAIL_BODY = "This is a test email sent using Fireworks AI and Klavis integration."
EMAIL_RECIPIENT = "recipient@example.com"  # Replace with your email

# After OAuth authorization, create an agent with Gmail MCP server
agent = Agent(fireworks_client, klavis_client, gmail_mcp_instance.server_url)

# Send the email
response = agent.process_request(
    f"Send an email to {EMAIL_RECIPIENT} with subject {EMAIL_SUBJECT} and body {EMAIL_BODY}"
)

print(response)