File size: 4,127 Bytes
a4cd24c
 
 
c453d15
 
a4cd24c
2dada65
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a4cd24c
c453d15
 
a4cd24c
c453d15
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a4cd24c
 
 
 
 
 
 
c453d15
a4cd24c
c453d15
 
 
 
 
a4cd24c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
# filename: request_router.py

import gradio as gr
import os
import importlib.util
from typing import Any, Dict
import requests

# This is the only hardcoded Tool endpoint
def register_tool(tool_data):
    # Send a POST request to register the tool
    response = requests.post("https://huggingface.co/chat/tools/new", json=tool_data)
    return response.json()

def create_tool(tool_name, tool_data):
    """One tool to create them all.
        a AI model can wield this toolto create a new 'tool' script in the tools folder
        besides the values for the registration payload,
        the 'tool_data' should also be a string value 'tfn' which contains the code for the tool function script
        """
    print("processing create_tool ")
    print(tool_name)
    print(dir(tool_data))
    # set the payload for restration
    tool_payload = {
        "displayName": tool_data.get("displayName"), # str display name of the tool
        "description": tool_data.get("description"), # str dscription
        "color": tool_data.get("color"),  # Example color purple,red,green
        "icon": tool_data.get("icon"), # image binairy
        "baseUrl": "K00B404/toolshed", # always to the toolshed
        "endpoint": "/router", # always to the router
        "name": tool_data.get("name"), # str name for matching function
        "inputs":tool_data.get("inputs"), # List(str) list the tools input values
        "outputComponent": tool_data.get("outputComponent"),  # Adjust based on your tool's output
        "outputComponentIdx": tool_data.get("outputComponentIdx"), # Adjust based on your tool's output
        "showOutput": tool_data.get("showOutput") # bool
    }
    print(dir(tool_payload))
    
    # write the new tool script from tfn
    with open(f'./tools/{tool_data.get("name")}.py' 'w') as f:
              f.write(tool_data.get("tfn"))
    # send the registation request for the tool      
    return register_tool(tool_payload)


# Define the path to the tools folder
TOOLS_DIR = './tools'

# Function to dynamically load methods from files in the ./tools folder
def load_methods_from_tools():
    method_mapping = {}
    
    # List all .py files in the tools directory
    for filename in os.listdir(TOOLS_DIR):
        if filename.endswith('.py'):
            method_name = filename[:-3].lower()  # Remove '.py' and convert to lowercase
            
            # Load the module
            module_path = os.path.join(TOOLS_DIR, filename)
            spec = importlib.util.spec_from_file_location(method_name, module_path)
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

            # Map all functions from the module to the method mapping
            for attr in dir(module):
                if callable(getattr(module, attr)) and not attr.startswith("__"):
                    method_mapping[method_name] = getattr(module, attr)

    return method_mapping

# Load the method mapping at the start
method_mapping = load_methods_from_tools()

# Main request router method
def request_router(name: str, input_data: Dict[str, Any]) -> Dict[str, Any]:
    # Retrieve the appropriate method based on the name
    method = method_mapping.get(name)
    if method is None:
        return {"error": "Method not found"}
    
    # Call the method with the provided input data
    try:
        output = method(input_data)
        return output
    except Exception as e:
        return {"error": str(e)}

# Gradio Interface
def launch_gradio_app():
    with gr.Blocks() as demo:
        gr.Markdown("# Request Router API")
        with gr.Row():
            name_input = gr.Textbox(label="Method Name")
            input_data_input = gr.Textbox(label="Input Data (JSON format)")
            output_output = gr.JSON(label="Output")
        
        submit_button = gr.Button("Submit")
        submit_button.click(
            fn=lambda name, input_data: request_router(name, eval(input_data)),
            inputs=[name_input, input_data_input],
            outputs=output_output
        )

    demo.launch()

if __name__ == "__main__":
    launch_gradio_app()