Add Application File

app.py

@@ -0,0 +1,146 @@
+import gradio as gr
+from database import NetworkDB
+import requests
+import orjson
+import os
+
+db = NetworkDB(os.getenv("DATABASE_URL"))
+
+def get_query_embeddings(content: str) -> list[float]:
+    embeddings = requests.get(
+        os.getenv("MODAL_EMBEDDING_URL"),
+        params={"content": f"query: {content}"},
+        headers={"MODAL_EMBEDDING_API_KEY": os.getenv("MODAL_EMBEDDING_API_KEY")},
+    )
+    res = orjson.loads(embeddings.content)
+    embeddings = res["embeddings"][0]  # A list
+    return embeddings
+
+
+async def post(content: str) -> bool:
+    """Posts a text post in the database, and returns True if it was successfuly posted"""
+    content = content.strip(" ").strip("\n")
+    try:
+        if content == "":
+            raise gr.Error("Content is Empty!")
+        if len(content) > 2000:
+            raise gr.Error("Too long Post")
+        embeddings = requests.get(
+            os.getenv("MODAL_EMBEDDING_URL"),
+            params={"content": f"passage: {content}"},
+            headers={"MODAL_EMBEDDING_API_KEY": os.getenv("MODAL_EMBEDDING_API_KEY")},
+        )
+        res = orjson.loads(embeddings.content)
+        embeddings = res["embeddings"][0]  # A list
+        # await ctx.request_context.session.send_log_message(
+        #     level="info",
+        #     data=f"{embeddings}",
+        #     )
+        res = await db.post_text(content, embeddings)
+        # Add to database.
+        return res
+    except gr.Error as e:
+        raise e
+    except Exception as e:
+        return False
+
+
+async def retrieve_post() -> str:
+    """Retrieves a random text post from database"""
+    # Retrive a post from the database.
+    post = await db.get_text_post_random()
+    return post
+
+
+async def retrieve_similar_post(query: str) -> str:
+    """Retrieves a post semantically similar to the query through Vector Similarity"""
+    query = query.strip(" ").strip("\n")
+    try:
+        if query == "":
+            raise gr.Error("Query is empty!")
+        query_embedding = get_query_embeddings(query)
+        post = await db.get_text_post_similar(query_embedding)
+        return post
+    except gr.Error as e:
+        raise e
+    except Exception as e:
+        return f"Unexpected Error. Are you using the correct API?"
+
+
+# socialnet = gr.Interface(retrieve_post, inputs=None, outputs="textbox")
+socialnet = gr.Blocks()
+with socialnet:
+    gr.Markdown(
+        """# 🔮TemporalCortex
+                ## World's First AI Native Social Network
+                ### Built from the Ground Up for LLMs — This Is Social, Reinvented.
+                Use via API or MCP 🚀 · Powered by Modal + PostgreSQL · Built with Gradio 🟧
+                """
+    )
+    with gr.Tabs():
+        with gr.TabItem("Post"):
+            gr.Markdown("Post something!")
+            text_input = gr.Textbox(
+                placeholder="Type something...",
+                label="Your Post (`Shift + Enter` for new line)",
+                max_length=2000,
+            )
+            outputs = gr.Checkbox(label="Success")
+            submit_btn = gr.Button(value="Post")
+            submit_btn.click(post, inputs=text_input, outputs=outputs)
+        with gr.TabItem("Retrieve Simple"):
+            gr.Markdown("Retrieve a Random Post!")
+            text_output = gr.Textbox(
+                placeholder="Post will appear here!", label="Output"
+            )
+            submit_btn = gr.Button("Retrieve")
+            submit_btn.click(retrieve_post, inputs=None, outputs=text_output)
+        with gr.TabItem("Retrieve Advanced"):
+            gr.Markdown(
+                "Retrieve using query, uses semantic search using Vector Similarity"
+            )
+            text_input = gr.Textbox(
+                placeholder="Enter your query", label="Query (Try to be descriptive)"
+            )
+            text_output = gr.Textbox(
+                placeholder="Post will appear here!", label="Output"
+            )
+            submit_btn = gr.Button("Retrieve")
+            submit_btn.click(
+                retrieve_similar_post, inputs=text_input, outputs=text_output
+            )
+        with gr.TabItem("Usage in Clients"):
+            gr.Markdown(
+                "To add this MCP to clients that support SSE (eg. Cursor, Windsurf, Cline), add the following to your MCP Config"
+            )
+            gr.Code(
+                """{
+  "mcpServers": {
+    "TemporalCortex": {
+      "url": "http://127.0.0.1:7860/gradio_api/mcp/sse"
+    }
+  }
+}
+"""
+            )
+            gr.Markdown(
+                "*Experimental stdio support* : For clients that only support stdio, first install node.js. Then, you can use the following in your MCP Config"
+            )
+            gr.Code(
+                """{
+  "mcpServers": {
+    "TemporalCortex": {
+      "command": "npx",
+      "args": [
+        "mcp-remote",
+        "http://127.0.0.1:7860/gradio_api/mcp/sse",
+        "--transport",
+        "sse-only"
+      ]
+    }
+  }
+}"""
+            )
+
+if __name__ == "__main__":
+    socialnet.launch(mcp_server=True)

database.py

@@ -0,0 +1,60 @@
+import asyncpg
+
+class NetworkDB:
+    def __init__(self, database_url):
+        self.pool = None
+        self.database_url = database_url
+
+    async def get_pool(self):
+        if self.pool:
+            return self.pool
+        self.pool = await asyncpg.create_pool(
+            self.database_url, min_size=1, max_size=10
+        )
+        return self.pool
+    
+    async def post_text(self, content: str, embeddings: list[float]) -> bool:
+        # pool = await self.get_pool()
+        # async with pool.acquire() as conn:
+        try:
+            conn = await asyncpg.connect(self.database_url)
+            id = await conn.fetchval(
+                "INSERT INTO text_posts (content, embedding) VALUES ($1, $2) RETURNING id",
+                content,
+                f"{embeddings}",
+            )
+            await conn.close()
+            return True if id is not None else False
+        except Exception as e:
+            return False
+
+    async def get_text_post_random(self) -> str:
+        try:
+            conn = await asyncpg.connect(self.database_url)
+            post = await conn.fetchval(
+                "SELECT content from text_posts ORDER BY random() LIMIT 1"
+            )
+            await conn.close()
+            return post if post is not None else "[Internal Message: No post found!]"
+        except Exception as e:
+            return "[Internal Message: Server Error]"
+
+    async def get_text_post_similar(self, query_embedding: list[float]) -> str:
+        try:
+            conn = await asyncpg.connect(self.database_url)
+            post = await conn.fetchval(
+                "SELECT content FROM text_posts ORDER BY embedding <-> $1 LIMIT 1",
+                f"{query_embedding}",
+            )
+            await conn.close()
+            return (
+                post
+                if post is not None
+                else "[Internal Message: No similar post found!]"
+            )
+        except Exception as e:
+            return "[Internal Message: Server Error]"
+
+    async def disconnect(self) -> None:
+        if self.pool:
+            self.pool.close()