tools.py

# tools.py
import os
import subprocess
import requests
from langchain.tools import tool
import json
from duckduckgo_search import DDGS

@tool
def file_operations(input_str: str) -> str:
    """Perform file operations: read, write, or list files. Input is a JSON string with 'action', 'file_path', and optional 'content'."""
    try:
        # Parse JSON input
        input_data = json.loads(input_str)
        action = input_data.get("action")
        file_path = input_data.get("file_path")
        content = input_data.get("content")

        if action == "read" and file_path:
            with open(file_path, "r") as f:
                return f.read()
        elif action == "write" and file_path and content:
            with open(file_path, "w") as f:
                f.write(content)
                return f"Wrote to {file_path}"
        elif action == "list" and file_path:
            files = os.listdir(file_path)
            return ", ".join(files)
        else:
            return "Invalid action or parameters"
    except Exception as e:
        return f"Error: {str(e)}"

@tool
def execute_code(code: str) -> str:
    """Execute Python code and return the output."""
    try:
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True,
            text=True,
            timeout=10
        )
        return result.stdout or result.stderr
    except Exception as e:
        return f"Error: {str(e)}"

@tool
def web_research(query: str) -> str:
    """Search the web using DuckDuckGo and summarize results."""
    try:
        with DDGS() as ddgs:
            results = ddgs.text(query, max_results=3)
        if not results:
            return "No results found"
        snippets = [result['body'] for result in results if 'body' in result]
        return "\n".join(snippets) or "No snippets available"
    except Exception as e:
        return f"Error: {str(e)}"

@tool
def mcp_request(data: str) -> str:
    """Send a request to an MCP server."""
    try:
        with open("config.json", "r") as f:
            config = json.load(f)["mcp_server"]
        response = requests.post(
            config["url"] + config["endpoint"],
            headers={"Authorization": f"Bearer {config['api_key']}"},
            json={"input": data}
        )
        return response.json().get("output", "No output")
    except Exception as e:
        return f"Error: {str(e)}"

tools = [file_operations, execute_code, web_research, mcp_request]