initial commit with gradio app

.gitignore

@@ -0,0 +1,3 @@
+venv/
+tools/__pycache__
+.env

app.py

@@ -0,0 +1,44 @@
+import gradio as gr
+from research import research
+from textblob import TextBlob
+
+def research_query(query):
+    """
+    Function to handle research queries through Gradio interface
+
+    Args:
+        text (str): The query to perform websearch and provide summary.
+    Returns:
+        text (str): A detailed summary on the query asked by perfoming web search.
+    """
+    if not query.strip():
+        return "Please enter a valid query"
+    
+    try:
+        result = research(query)
+        return result
+    except Exception as e:
+        return f"Error processing query: {str(e)}"
+
+# Create Gradio interface
+demo = gr.Interface(
+    fn=research_query,
+    inputs=gr.Textbox(
+        lines=3, 
+        placeholder="Enter your research query here...",
+        label="Research Query"
+    ),
+    outputs=gr.Textbox(
+        lines=10,
+        label="Research Results"
+    ),
+    title="Research Assistant",
+    description="Enter a query to get detailed research results using ReAct agent.",
+    examples=[
+        ["What are the latest developments in quantum computing?"],
+        ["Explain the impact of artificial intelligence on healthcare"],
+    ]
+)
+
+if __name__ == "__main__":
+    demo.launch(mcp_server=True)

requirements.txt

@@ -0,0 +1,8 @@
+requests
+openai
+python-dotenv
+markdownify
+mcp[cli]
+httpx
+gradio[mcp]
+textblob

research.py

@@ -0,0 +1,178 @@
+import os
+from typing import List, Dict, Any, Optional
+from openai import OpenAI
+import json
+from tools import SearchTool, FetchTool, SummarizeTool
+from dotenv import load_dotenv
+import httpx
+from mcp.server.fastmcp import FastMCP
+from openai.types.chat import ChatCompletionMessage
+from openai.types.chat.chat_completion import ChatCompletion
+
+# mcp = FastMCP("researcher")
+
+load_dotenv()
+
+class ReActAgent:
+    def __init__(self, client):
+        self.client = client
+        self.model = "qwen-3-32b"
+        self.conversation_history: List[Dict[str, str]] = []
+        self.max_history_length = 10  # Limit conversation history
+        self.tools = [
+            SearchTool(),
+            FetchTool(),
+            SummarizeTool()
+        ]
+
+        self.tools_json = [
+            {
+                "type": "function",
+                "function": tool.to_json()
+            }
+            for tool in self.tools
+        ]
+        self.tools_map = {tool.name: tool for tool in self.tools}
+        self.process_log = []  # Store the intermediate process
+
+    def _execute_tool(self, tool_call: Dict[str, Any]) -> str:
+        """Execute the called tool and return the result."""
+        try:
+            tool_name = tool_call.function.name
+            arguments = json.loads(tool_call.function.arguments)
+
+            if tool_name not in self.tools_map:
+                return f"Error: Unknown tool: {tool_name}"
+            
+            tool = self.tools_map[tool_name]
+            result = tool(**arguments)
+            
+            # Log the tool execution
+            self.process_log.append({
+                "tool": tool_name,
+                "arguments": arguments,
+                "result": result
+            })
+            
+            return result
+        except json.JSONDecodeError:
+            error_msg = "Error: Invalid tool arguments format"
+            self.process_log.append({
+                "tool": tool_call.function.name,
+                "arguments": tool_call.function.arguments,
+                "result": error_msg
+            })
+            return error_msg
+        except Exception as e:
+            error_msg = f"Error executing tool: {str(e)}"
+            self.process_log.append({
+                "tool": tool_call.function.name,
+                "arguments": tool_call.function.arguments,
+                "result": error_msg
+            })
+            return error_msg
+
+    def _truncate_history(self):
+        """Keep only the most recent messages to prevent context overflow."""
+        if len(self.conversation_history) > self.max_history_length:
+            self.conversation_history = self.conversation_history[-self.max_history_length:]
+
+    def _format_process_log(self) -> str:
+        """Format the process log into a readable string."""
+        if not self.process_log:
+            return "No intermediate steps were taken."
+            
+        formatted_log = ["<intermediate_steps>"]
+        for i, step in enumerate(self.process_log, 1):
+            formatted_log.append(f"\nStep {i}:")
+            formatted_log.append(f"Tool: {step['tool']}")
+            formatted_log.append(f"Arguments: {json.dumps(step['arguments'], indent=2)}")
+            formatted_log.append(f"Result: {step['result']}")
+        formatted_log.append("</intermediate_steps>")
+        return "\n".join(formatted_log)
+
+    def run(self, user_input: str) -> str:
+        """Run the ReAct loop for a single user input."""
+        if not user_input or not isinstance(user_input, str):
+            return "Error: Invalid input. Please provide a valid string query."
+
+        try:
+            # Reset process log for new query
+            self.process_log = []
+            
+            # Add user input to conversation history
+            self.conversation_history.append({"role": "user", "content": user_input})
+            print(f"\n\nUser input: {user_input}\n--------------------------------\n")
+            
+            while True:
+                try:
+                    # Get response from the model
+                    response: ChatCompletion = self.client.chat.completions.create(
+                        model=self.model,
+                        messages=self.conversation_history,
+                        tools=self.tools_json,
+                    )
+                    
+                    message: ChatCompletionMessage = response.choices[0].message
+                    
+                    # Add assistant's response to conversation history
+                    self.conversation_history.append({
+                        "role": "assistant",
+                        "content": message.content if message.content else "",
+                        "tool_calls": message.tool_calls
+                    })
+                    
+                    # If no tool calls, return the response with process log
+                    if not message.tool_calls:
+                        print("No tool calls\nExiting loop\n--------------------------------")
+                        final_response = message.content or "No response generated"
+                        process_log = self._format_process_log()
+                        return f"{process_log}\n\n{final_response}"
+                    
+                    # Execute the tool calls
+                    tool_results = []
+                    for tool_call in message.tool_calls:
+                        print(f"Tool call: {tool_call.function.name}\nTool arguments: {tool_call.function.arguments}")
+                        tool_result = self._execute_tool(tool_call)
+                        print(f"Tool result: {tool_result}\n--------------------------------\n")
+                        tool_results.append({
+                            "tool_call_id": tool_call.id,
+                            "role": "tool",
+                            "name": tool_call.function.name,
+                            "content": tool_result
+                        })
+                    
+                    # Add tool results to conversation history
+                    self.conversation_history.extend(tool_results)
+                    self._truncate_history()
+
+                except Exception as e:
+                    error_msg = f"Error during model interaction: {str(e)}"
+                    process_log = self._format_process_log()
+                    return f"{error_msg}\n\n{process_log}"
+
+        except Exception as e:
+            error_msg = f"Error in research process: {str(e)}"
+            process_log = self._format_process_log()
+            return f"{error_msg}\n\n{process_log}"
+
+# @mcp.tool()
+def research(query: str) -> str:
+    """Get final answer on the query after detailed research"""
+    try:
+        api_key = os.environ.get("CEREBRAS_API_KEY")
+        if not api_key:
+            return "Error: Please set CEREBRAS_API_KEY environment variable"
+        
+        client = OpenAI(
+            base_url="https://api.cerebras.ai/v1",
+            api_key=api_key
+        )
+        
+        agent = ReActAgent(client)
+        return agent.run(query)
+    except Exception as e:
+        return f"Error in research function: {str(e)}"
+
+# if __name__ == "__main__":
+#     mcp.run()

tools/__init__.py

@@ -0,0 +1,6 @@
+from .search import SearchTool
+from .fetch import FetchTool
+from .summarize import SummarizeTool
+from .tool import Tool
+
+__all__ = ["SearchTool", "FetchTool", "SummarizeTool", "Tool"]

tools/fetch.py

@@ -0,0 +1,31 @@
+from .tool import Tool
+from markdownify import markdownify
+import requests
+
+class FetchTool(Tool):
+    def __init__(self):
+        super().__init__(
+            name="fetch",
+            description="Fetch the content of a URL and return the markdownified version of the content",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "url": {"type": "string", "description": "The URL to fetch"}
+                }
+            }
+        )
+    
+    def __call__(self, url: str):
+        try:
+            if not url:
+                return "Error: URL parameter is required"
+                
+            resp = requests.get(url)
+            resp.raise_for_status()  # Raise an exception for bad status codes
+            
+            return markdownify(resp.text)
+            
+        except requests.exceptions.RequestException as e:
+            return f"Error fetching URL: {str(e)}"
+        except Exception as e:
+            return f"Unexpected error while processing URL: {str(e)}"

tools/search.py

@@ -0,0 +1,65 @@
+import requests
+from dotenv import load_dotenv
+import os
+from .tool import Tool
+
+load_dotenv("./.env")
+
+class SearchTool(Tool):
+    def __init__(self):
+        super().__init__(
+            name="search",
+            description="Search the web for information",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "query": {"type": "string", "description": "The search query"}
+                }
+            }
+        )
+        
+        self.api_key = os.environ.get("GOOGLE_API_KEY")
+        self.search_engine_id = os.environ.get("GOOGLE_CSE_ID")
+        
+        if not self.api_key:
+            raise ValueError("Please set GOOGLE_API_KEY environment variable")
+        if not self.search_engine_id:
+            raise ValueError("Please set GOOGLE_CSE_ID environment variable")
+    
+    def __call__(self, query: str):
+        try:
+            if not query:
+                return "Error: Query parameter is required"
+                
+            params = {
+                "q": query,
+                "key": self.api_key,
+                "cx": self.search_engine_id
+            }
+            
+            resp = requests.get("https://www.googleapis.com/customsearch/v1", params=params)
+            resp.raise_for_status()  # Raise an exception for bad status codes
+            
+            _results = resp.json().get("items", [])
+            results = []
+            for result in _results[:3]:
+                results.append({
+                    "title": result.get("title", "No title"),
+                    "link": result.get("link", "No link"),
+                    "snippet": result.get("snippet", "No snippet")
+                })
+            
+            if not results:
+                return "No results found for the given query."
+            
+            # Format results as a string
+            formatted_results = []
+            for i, result in enumerate(results, 1):
+                formatted_results.append(f"Result {i}:\nTitle: {result['title']}\nLink: {result['link']}\nSnippet: {result['snippet']}\n")
+            
+            return "\n".join(formatted_results)
+            
+        except requests.exceptions.RequestException as e:
+            return f"Error during search: {str(e)}"
+        except Exception as e:
+            return f"Unexpected error during search: {str(e)}"

tools/summarize.py

@@ -0,0 +1,42 @@
+from .tool import Tool
+from openai import OpenAI
+from dotenv import load_dotenv
+import os
+
+load_dotenv("./.env")
+
+class SummarizeTool(Tool):
+    def __init__(self):
+        super().__init__(
+            name="summarize",
+            description="Summarize the content of a URL",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "content": {"type": "string", "description": "The content to summarize"}
+                }
+            }
+        )
+        
+        api_key = os.environ.get("CEREBRAS_API_KEY")
+        if not api_key:
+            raise ValueError("Please set CEREBRAS_API_KEY environment variable")
+            
+        self.client = OpenAI(base_url="https://api.cerebras.ai/v1", api_key=api_key)
+    
+    def __call__(self, **kwargs):
+        try:
+            content = kwargs.get("content")
+            if not content:
+                return "Error: Content parameter is required"
+                
+            response = self.client.chat.completions.create(
+                model="qwen-3-32b",
+                messages=[
+                    {"role": "system", "content": "You are a helpful assistant that summarizes content while keeping the all important information."},
+                    {"role": "user", "content": content}
+                ]
+            )
+            return response.choices[0].message.content
+        except Exception as e:
+            return f"Error during summarization: {str(e)}"

tools/tool.py

@@ -0,0 +1,15 @@
+class Tool:
+    def __init__(self, name: str, description: str, inputSchema: dict):
+        self.name = name
+        self.description = description
+        self.inputSchema = inputSchema
+    
+    def __repr__(self):
+        return f"Tool(name={self.name}, description={self.description}, inputSchema={self.inputSchema})"
+    
+    def to_json(self):
+        return {
+            "name": self.name,
+            "description": self.description,
+            "parameters": self.inputSchema
+        }