app.py

import gradio as gr

import config
import utils


# Import obb (don't reload it)
if gr.NO_RELOAD:
    from openbb import obb

# A set of submodules to include when searching for tools, registering
# them to the MCP server, and dynamically building the Gradio interface
# List of valid values: https://docs.openbb.co/platform/reference
INCLUDE = {"currency", "equity", "news"}

# Get tool names and guides
print("Getting tool names and guides...")
ARGS = "obb", obb # type: ignore
tool_names = utils.get_callable_names(*ARGS, include=INCLUDE)
tool_guides = utils.generate_callable_guides(*ARGS, tool_names)


# Build the demo interface
print("Building the demo...")
with gr.Blocks() as demo:
    gr.Markdown("# OpenBB MCP")

    # Reference guide
    with gr.Accordion("Tool Reference Guide 📃", open=False):
        for tool_name, tool_guide in zip(tool_names, tool_guides):
            
            # Create a collapsible for each tool
            with gr.Accordion(f"{tool_name}", open=False):
                gr.Markdown(f"```\n{tool_guide}```")

                # Dynamically generate tool test UI
                tool_params = utils.get_callable_params(*ARGS, tool_name)
                tool_param_inputs = []
                tool_param_kinds = []
                with gr.Row():
                    for param in tool_params:
                        tool_param_inputs.append(
                            gr.Textbox(
                              label=param["name"],
                              value=str(param["default"]) if param["default"] else ""
                            )
                        )
                        tool_param_kinds.append(param["kind"])
                output = gr.Textbox(label="Output", lines=10)
                test_btn = gr.Button("Run")

                # Format vars for dynamic function generation
                t = tool_name
                tool_name = tool_name.replace(".", "_")
                tool_guide = tool_guide.replace(t, tool_name)
                tool_param_names = [
                    param["name"] for param in tool_params
                ]
                csv_tool_names = ", ".join(tool_param_names)

                # Dynamically generate function
                # TODO: Fragile. Refactor if possible
                namespace = { # Create a local namespace
                    "utils": utils,
                    "ARGS": ARGS,
                    "tool_param_kinds": tool_param_kinds
                }
                exec(f"""\
import inspect
import json               

def {tool_name}({csv_tool_names}) -> str:
    \"\"\"
    {tool_guide}
    \"\"\"
    print("{tool_name}:", locals())
    args = []
    try:
        kwargs = json.loads(kwargs)
    except Exception:
        kwargs = {{}}
    for kind, name, value in zip(
        tool_param_kinds,
        {tool_param_names},
        [{csv_tool_names}],
    ):
        if kind in {{
            inspect.Parameter.POSITIONAL_ONLY,
            inspect.Parameter.POSITIONAL_OR_KEYWORD,
        }}:
            args.append(value if value else None)
    return utils.test_callable(*ARGS, "{tool_name}", *args, **kwargs)
""", namespace)

                # Extract tool from namespace and register to the MCP server
                test_btn.click(
                    fn=namespace[tool_name], # type: ignore
                    inputs=tool_param_inputs,
                    outputs=output,
                )

    # Usage instructions
    with gr.Accordion("Use via MCP 🛠️", open=False):
        gr.Markdown("""\
**SSE support**: To add this MCP to clients that support SSE (e.g. Cursor, Windsurf, Cline), simply add the following configuration to your MCP config:
 
```
{
  "mcpServers": {
    "OpenBB-MCP": {
      "url": "http://xarical-openbb-mcp.hf.space/gradio_api/mcp/sse"
    }
  }
}
```

**Stdio support**: For clients that only support stdio, first install Node.js. Then, you can use the following command:
 
```
{
  "mcpServers": {
    "OpenBB-MCP": {
      "command": "npx",
      "args": [
        "mcp-remote",
        "http://xarical-openbb-mcp.hf.space/gradio_api/mcp/sse",
        "--transport",
        "sse-only"
      ]
    }
  }
}
```
""")


# Launch the demo
if __name__ == "__main__":
    print("Starting the demo...")
    demo.launch(
        server_port=7860,
        show_api=False,
        mcp_server=True,
    )