Add app.py, utils.py, and config.py

app.py

@@ -0,0 +1,137 @@
+import gradio as gr
+
+import utils
+
+
+# Import obb (don't reload it)
+if gr.NO_RELOAD:
+    from openbb import obb
+
+# Get tool names and guides
+ARGS = "obb", obb # type: ignore
+tool_names = utils.get_callable_names(
+    *ARGS, 
+    include={"equity"}
+)
+tool_guides = utils.generate_callable_guides(*ARGS, tool_names)
+
+
+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
+                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(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"
+      ]
+    }
+  }
+}
+```
+""")
+
+
+if __name__ == "__main__":
+    demo.launch(
+        server_name="localhost",
+        server_port=7860,
+        show_api=True,
+        mcp_server=True,
+    )

config.py

@@ -0,0 +1,11 @@
+from openbb import obb
+import openbb
+
+
+# Configure LLM mode for OpenBB
+obb.user.preferences.output_type="llm"
+obb.system.python_settings.docstring_sections=['description', 'examples']
+obb.system.python_settings.docstring_max_length=1024
+
+# Build OpenBB
+openbb.build()

requirements.txt

@@ -0,0 +1,2 @@
+gradio[mcp]
+openbb

utils.py

@@ -0,0 +1,124 @@
+import inspect
+from typing import Any, Optional
+
+
+def get_nested_attr(
+    module: ..., 
+    attr_path: str,
+) -> Any | None:
+    """Loop through nested attributes."""
+    x = module
+    for attr in attr_path.split("."):
+        x = getattr(x, attr, None)
+    return x
+
+
+def get_callable_names(
+    prefix: str, 
+    module: ..., 
+    max_depth: int = 3, 
+    depth: int = 0,
+    include: Optional[set[str]] = None,
+) -> list[str]:
+    """
+    Recursively retrieve the names of all callables in a given module.
+    """
+    callable_names = []
+    if depth > max_depth:
+        return callable_names
+    for attr in dir(module):
+        if (
+            attr.startswith("_") or
+            attr in {"model_fields", "model_computed_fields"} or
+            include and attr not in include
+        ):
+            continue
+        try:
+            value = getattr(module, attr)
+        except Exception:
+            continue
+        full_name = f"{prefix}.{attr}"
+        if callable(value):
+            callable_names.append(full_name)
+        elif hasattr(value, "__dict__") or hasattr(value, "__module__"):
+            callable_names.extend(
+                get_callable_names(
+                    full_name,
+                    value,
+                    max_depth,
+                    depth + 1
+                ) # Recurse
+            )
+    return callable_names
+
+
+def get_callable_params(
+    prefix: str, 
+    module: ...,
+    callable_name: str
+) -> list[dict[str, str]]:
+    """
+    Get parameter names and default values for a given callable.
+    """
+    callable = get_nested_attr(
+        module, 
+        callable_name.replace(f"{prefix}.", "", 1)
+    )
+    if not callable:
+        return []
+    try:
+        sig = inspect.signature(callable)
+        params = []
+        for name, param in sig.parameters.items():
+            if name == "self":
+                continue
+            default = (
+                param.default if
+                param.default is not inspect.Parameter.empty else 
+                ""
+            )
+            kind = param.kind
+            params.append({
+                "name": name,
+                "default": default, 
+                "kind": kind
+            })
+        return params
+    except Exception:
+        return []
+
+
+def test_callable(
+    prefix: str,
+    module: ...,
+    callable_name: str,
+    *args, **kwargs
+) -> str:
+    """Test a given callable and return the output."""
+
+    callable = get_nested_attr(
+        module,
+        callable_name.replace("_", ".").replace(f"{prefix}.", "", 1)
+    )
+    if not callable:
+        return f"Callable '{callable_name}' not found."
+    try:
+        result = callable(*args, **kwargs)
+        return result
+    except TypeError as e:
+        return f"Callable '{callable_name}' requires arguments: {e}"
+    except Exception as e:
+        return f"Error running '{callable_name}': {e}"
+
+
+def generate_callable_guides(prefix: str, module: ..., callable_names: list[str]) -> list[str]:
+    """
+    Generate callable reference guides from docstrings, parameters, and
+    parameter/return types.
+    """
+    guide = []
+    for callable in callable_names:
+        callable = get_nested_attr(module, callable.replace(f"{prefix}.", "", 1))
+        doc = callable.__doc__ if callable and hasattr(callable, "__doc__") else "No docstring available."
+        guide.append(f"{doc}")
+    return guide