OpenBB agent copilot

openbb-agent-copilot/README.md

@@ -0,0 +1,44 @@
+# OpenBB Agent Copilot
+
+This example provides a copilot that leverages the OpenBB Platform to retrieve financial data.
+
+## Overview
+
+This implementation utilizes a FastAPI application to serve as the backend for
+the copilot.
+
+## Getting started
+
+Here's how to get your copilot up and running:
+
+### Prerequisites
+
+Ensure you have poetry, a tool for dependency management and packaging in
+Python, installed as well as your OpenAI API key.
+
+### Installation and Running
+
+1. Clone this repository to your local machine.
+2. Set the OpenAI API key as an environment variable in your .bashrc or .zshrc file:
+
+``` sh
+# in .zshrc or .bashrc
+export OPENAI_API_KEY=<your-api-key>
+export OPENBB_PAT=<your-openbb-pat>
+```
+
+The latter can be found here: https://my.openbb.co/app/platform/pat
+
+3. Install the necessary dependencies:
+
+``` sh
+poetry install --no-root
+```
+
+4.Start the API server:
+
+``` sh
+poetry run uvicorn openbb_agent_copilot.main:app --port 7777 --reload
+```
+
+This command runs the FastAPI application, making it accessible on your network.

openbb-agent-copilot/openbb_agent_copilot/copilots.json

@@ -0,0 +1,13 @@
+{
+	"example_copilot": {
+		"name": "OpenBB Agent Copilot",
+		"description": "AI financial analyst using the OpenBB Platform.",
+		"image": "https://github.com/user-attachments/assets/010d7590-0a65-4b3f-b21a-0cbc0d95bcb9",
+		"hasStreaming": true,
+		"hasDocuments": false,
+		"hasFunctionCalling": false,
+		"endpoints": {
+			"query": "http://localhost:7777/v1/query"
+		}
+	}
+}

openbb-agent-copilot/openbb_agent_copilot/main.py

@@ -0,0 +1,83 @@
+import re
+import json
+from pathlib import Path
+from typing import AsyncGenerator
+from openbb_agents.agent import openbb_agent
+import os
+
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from magentic import (
+    UserMessage,
+    AssistantMessage,
+)
+
+from dotenv import load_dotenv
+from sse_starlette import EventSourceResponse
+from .models import AgentQueryRequest
+
+load_dotenv(".env")
+app = FastAPI()
+
+origins = [
+    "http://localhost",
+    "http://localhost:1420",
+    "http://localhost:5050",
+    "https://pro.openbb.dev",
+    "https://pro.openbb.co",
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+def sanitize_message(message: str) -> str:
+    """Sanitize a message by escaping forbidden characters."""
+    cleaned_message = re.sub(r"(?<!\{)\{(?!{)", "{{", message)
+    cleaned_message = re.sub(r"(?<!\})\}(?!})", "}}", cleaned_message)
+    return cleaned_message
+
+
+async def create_message_stream(
+    content: str,
+) -> AsyncGenerator[dict, None]:
+    yield {"event": "copilotMessageChunk", "data": {"delta": content}}
+
+
+@app.get("/copilots.json")
+def get_copilot_description():
+    """Widgets configuration file for the OpenBB Terminal Pro"""
+    return JSONResponse(
+        content=json.load(open((Path(__file__).parent.resolve() / "copilots.json")))
+    )
+
+
+@app.post("/v1/query")
+async def query(request: AgentQueryRequest) -> EventSourceResponse:
+    """Query the Copilot."""
+
+    chat_messages: list[AssistantMessage | UserMessage] = []
+    for message in request.messages:
+        if message.role == "ai":
+            chat_messages.append(
+                AssistantMessage(content=sanitize_message(message.content))
+            )
+        elif message.role == "human":
+            chat_messages.append(UserMessage(content=sanitize_message(message.content)))
+
+    try:
+        result = openbb_agent(
+            str(chat_messages), verbose=True, openbb_pat=os.getenv("OPENBB_PAT")
+        )
+        return EventSourceResponse(
+            create_message_stream(result), media_type="text/event-stream"
+        )
+
+    except Exception as err:
+        raise HTTPException(status_code=400, detail=str(err))

openbb-agent-copilot/openbb_agent_copilot/models.py

@@ -0,0 +1,45 @@
+from typing import Any
+from uuid import UUID
+from pydantic import BaseModel, Field, field_validator
+from enum import Enum
+
+
+class RoleEnum(str, Enum):
+    ai = "ai"
+    human = "human"
+
+
+class LlmMessage(BaseModel):
+    role: RoleEnum = Field(
+        description="The role of the entity that is creating the message"
+    )
+    content: str = Field(description="The content of the message")
+
+
+class BaseContext(BaseModel):
+    uuid: UUID = Field(description="The UUID of the widget.")
+    name: str = Field(description="The name of the widget.")
+    description: str = Field(
+        description="A description of the data contained in the widget"
+    )
+    content: Any = Field(description="The data content of the widget")
+    metadata: dict[str, Any] | None = Field(
+        default=None,
+        description="Additional widget metadata (eg. the selected ticker, etc)",
+    )
+
+
+class AgentQueryRequest(BaseModel):
+    messages: list[LlmMessage] = Field(
+        description="A list of messages to submit to the copilot."
+    )
+    context: list[BaseContext] | None = Field(
+        default=None,
+        description="Additional context.",
+    )
+
+    @field_validator("messages")
+    def check_messages_not_empty(cls, value):
+        if not value:
+            raise ValueError("messages list cannot be empty.")
+        return value