Feat/potpie bundle

src/backend/base/langflow/components/potpie/__init__.py

@@ -0,0 +1,7 @@
+from .parse_repo import ParseRepository
+from .query import Query
+
+__all__ = [
+    "ParseRepository",
+    "Query",
+]

src/backend/base/langflow/components/potpie/parse_repo.py

@@ -0,0 +1,98 @@
+import time
+
+import httpx
+
+from langflow.custom import Component
+from langflow.inputs import MessageTextInput
+from langflow.io import SecretStrInput
+from langflow.schema import Message
+from langflow.template import Output
+
+
+class ParseRepository(Component):
+    display_name = "Parse Repository"
+    description = "Parses a github repository and stores the knowledge graph in potpie\
+                        (note: first time parsing might take a while)."
+    documentation = "https://docs.potpie.ai"
+    icon = "Potpie"
+
+    inputs = [
+        SecretStrInput(
+            name="potpie_api_key",
+            display_name="Potpie API Key",
+            info="API Key from app.potpie.ai",
+            required=True,
+        ),
+        MessageTextInput(
+            name="repo_name",
+            display_name="Repository Name",
+            info="From github (e.g. 'langflow-ai/langflow')",
+            required=True,
+        ),
+        MessageTextInput(
+            name="branch_name",
+            display_name="Branch Name",
+            info="Name of the branch to parse.",
+            placeholder="main",
+            required=True,
+        ),
+    ]
+
+    outputs = [
+        Output(display_name="Project ID", name="project_id", method="parse_repo"),
+    ]
+
+    def get_status(self, project_id: str) -> str:
+        endpoint = f"https://production-api.potpie.ai/api/v2/parsing-status/{project_id}"
+
+        headers = {
+            "X-API-Key": self.potpie_api_key,
+            "Content-Type": "application/json",
+        }
+
+        try:
+            response = httpx.get(endpoint, headers=headers, timeout=10000)
+            response.raise_for_status()
+
+            json = response.json()
+            status: str = json["status"]
+
+        except httpx.HTTPStatusError as e:
+            raise ValueError(e.response.text) from e
+
+        return status
+
+    def parse_repo(self) -> Message:
+        endpoint = "https://production-api.potpie.ai/api/v2/parse"
+
+        headers = {
+            "X-API-Key": self.potpie_api_key,
+            "Content-Type": "application/json",
+        }
+
+        payload = {
+            "repo_name": self.repo_name,
+            "branch_name": self.branch_name,
+        }
+
+        try:
+            response = httpx.post(endpoint, headers=headers, json=payload, timeout=10000)
+            response.raise_for_status()
+
+            json = response.json()
+            project_id: str = json["project_id"]
+
+        except httpx.HTTPStatusError as e:
+            raise ValueError(e.response.text) from e
+
+        # Wait for status change
+        status_res = "ready"
+
+        while True:
+            status = self.get_status(project_id)
+            if status in ("ready", "error"):
+                status_res = status
+                break
+            time.sleep(10)
+
+        return Message(text=project_id, status=status_res)

src/backend/base/langflow/components/potpie/query.py

@@ -0,0 +1,76 @@
+import httpx
+
+from langflow.custom import Component
+from langflow.inputs import DropdownInput, MessageTextInput, PromptInput
+from langflow.io import SecretStrInput
+from langflow.schema import Message
+from langflow.template import Output
+
+
+class Query(Component):
+    display_name = "Query Repository"
+    description = "Queries a parsed project with a context rich ai agent."
+    documentation = "https://docs.potpie.ai"
+    icon = "Potpie"
+
+    inputs = [
+        SecretStrInput(
+            name="potpie_api_key",
+            display_name="Potpie API Key",
+            info="API Key from app.potpie.ai",
+            required=True,
+        ),
+        MessageTextInput(
+            name="project_id",
+            display_name="ProjectID",
+            info="ID of the project from Potpie.",
+            required=True,
+        ),
+        DropdownInput(
+            name="agent_type",
+            display_name="Agent Type",
+            info="The type of agent to use for the query",
+            options=[
+                "codebase_qna_agent",
+                "debugging_agent",
+                "unit_test_agent",
+                "integration_test_agent",
+                "LLD_agent",
+                "code_changes_agent",
+                "code_generation_agent",
+            ],
+        ),
+        PromptInput(
+            name="query",
+            display_name="Query",
+            info="The message content for the agent to query the project with.",
+            placeholder="Example: What is the purpose of this project?",
+            required=True,
+        ),
+    ]
+
+    outputs = [
+        Output(display_name="Agent Response", name="agent_response", method="run_query"),
+    ]
+
+    def run_query(self) -> Message:
+        endpoint = f"https://conversation-production-api.potpie.ai/api/v2/project/{self.project_id}/message"
+        headers = {
+            "X-API-Key": self.potpie_api_key,
+            "Content-Type": "application/json",
+        }
+
+        payload = {
+            "content": self.query,
+            "agent_id": self.agent_type,
+        }
+
+        try:
+            response = httpx.post(endpoint, headers=headers, json=payload, timeout=10000)
+            response.raise_for_status()
+            res = response.json()
+            data = Message(text=res["message"])
+        except httpx.HTTPStatusError as e:
+            raise ValueError(e.response.text) from e
+
+        return data