feat(backend): Agents backend started

backend/.env.example

@@ -0,0 +1,16 @@
+# Use the testnet in development
+# ALEPH_API_URL=https://api.twentysix.testnet.network
+
+# Sender of the messages on Aleph
+ALEPH_SENDER=0x00
+# Private key of the address to send messages with
+ALEPH_SENDER_SK=
+# Public key of the address to send messages with
+ALEPH_SENDER_PK=
+# Channel on which to send messages
+ALEPH_CHANNEL=libertai
+# Type of the POST agent messages
+ALEPH_POST_TYPE=libertai-agent
+
+# Password used by the subscription backend for agent creation
+SUBSCRIPTION_BACKEND_PASSWORD=

backend/.gitignore

@@ -0,0 +1,5 @@
+/venv
+/dist
+/.idea
+
+.env

backend/README.md

@@ -0,0 +1,3 @@
+# LibertAI agents backend
+
+Small backend that handles agent creation and modification on [Aleph.im](https://aleph.im)

backend/pyproject.toml

@@ -0,0 +1,24 @@
+[tool.poetry]
+name = "libertai-agents-backend"
+version = "0.1.0"
+description = ""
+authors = ["LibertAI.io team <[email protected]>"]
+readme = "README.md"
+homepage = "https://libertai.io"
+repository = "https://github.com/LibertAI/libertai-agents"
+documentation = "https://docs.libertai.io"
+package-mode = false
+
+[tool.poetry.dependencies]
+python = "^3.12"
+fastapi = "^0.115.2"
+aleph-sdk-python = "^1.1.0"
+eciespy = "^0.4.2"
+
+[tool.poetry.group.dev.dependencies]
+mypy = "^1.12.0"
+ruff = "^0.7.0"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

backend/src/config.py

@@ -0,0 +1,32 @@
+import os
+
+from dotenv import load_dotenv
+
+
+class _Config:
+    ALEPH_API_URL: str | None
+
+    ALEPH_SENDER: str
+    ALEPH_SENDER_SK: bytes
+    ALEPH_SENDER_PK: bytes
+    ALEPH_CHANNEL: str
+    ALEPH_AGENT_POST_TYPE: str
+
+    SUBSCRIPTION_BACKEND_PASSWORD: str
+
+    def __init__(self):
+        load_dotenv()
+
+        self.ALEPH_API_URL = os.getenv("ALEPH_API_URL")
+        self.ALEPH_SENDER = os.getenv("ALEPH_SENDER")
+        self.ALEPH_SENDER_SK = os.getenv("ALEPH_SENDER_SK")  # type: ignore
+        self.ALEPH_SENDER_PK = os.getenv("ALEPH_SENDER_PK")  # type: ignore
+        self.ALEPH_CHANNEL = os.getenv("ALEPH_CHANNEL", "libertai")
+        self.ALEPH_AGENT_POST_TYPE = os.getenv(
+            "ALEPH_AGENT_POST_TYPE", "libertai-agent"
+        )
+
+        self.SUBSCRIPTION_BACKEND_PASSWORD = os.getenv("SUBSCRIPTION_BACKEND_PASSWORD")
+
+
+config = _Config()

backend/src/interfaces/agent.py

@@ -0,0 +1,38 @@
+from pydantic import BaseModel, validator
+
+from src.config import config
+from src.interfaces.subscription import SubscriptionAccount
+
+
+class DeleteAgentBody(BaseModel):
+    subscription_id: str
+    password: str
+
+    # noinspection PyMethodParameters
+    @validator("password")
+    def format_address(cls, password: str):
+        if password != config.SUBSCRIPTION_BACKEND_PASSWORD:
+            raise ValueError(
+                "Invalid password, you are not authorized to call this route"
+            )
+
+
+class SetupAgentBody(DeleteAgentBody):
+    account: SubscriptionAccount
+
+
+class UpdateAgentPutBody(BaseModel):
+    id: str
+    secret: str
+
+
+class Agent(BaseModel):
+    id: str
+    subscription_id: str
+    vm_hash: str | None
+    encrypted_secret: str
+    tags: list[str]
+
+
+class FetchedAgent(Agent):
+    post_hash: str

backend/src/interfaces/aleph.py

@@ -0,0 +1,8 @@
+from pydantic.main import BaseModel
+
+
+class AlephVolume(BaseModel):
+    comment: str
+    mount: str
+    ref: str
+    use_latest: bool

backend/src/interfaces/subscription.py

@@ -0,0 +1,16 @@
+# TODO: make a shared package for these types
+from enum import Enum
+
+from pydantic import BaseModel
+
+
+class SubscriptionChain(str, Enum):
+    base = "base"
+
+
+class SubscriptionAccount(BaseModel):
+    address: str
+    chain: SubscriptionChain
+
+    class Config:
+        schema_extra = {"example": {"address": "0x0000000000000000000000000000000000000000", "chain": "base"}}

backend/src/main.py

@@ -0,0 +1,141 @@
+from http import HTTPStatus
+from uuid import uuid4
+
+from aleph.sdk import AuthenticatedAlephHttpClient
+from aleph.sdk.chains.ethereum import ETHAccount
+from aleph_message.models.execution import Encoding
+from ecies import encrypt, decrypt
+from fastapi import FastAPI, HTTPException
+from starlette.datastructures import UploadFile
+from starlette.middleware.cors import CORSMiddleware
+
+from src.config import config
+from src.interfaces.agent import (
+    Agent,
+    UpdateAgentPutBody,
+    SetupAgentBody,
+    DeleteAgentBody,
+)
+from src.interfaces.aleph import AlephVolume
+from src.utils.agent import fetch_agents, fetch_agent_program_message
+from src.utils.storage import upload_file
+
+app = FastAPI(title="LibertAI subscriptions")
+
+origins = [
+    "https://chat.libertai.io",
+    "http://localhost:9000",
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.post("/agent", description="Setup a new agent on subscription")
+async def setup(body: SetupAgentBody) -> None:
+    agent_id = str(uuid4())
+
+    secret = str.encode(str(uuid4()), "utf-8")
+    # Encrypting the secret ID with our public key
+    encrypted_secret = encrypt(config.ALEPH_SENDER_PK, secret).decode()
+
+    agent = Agent(
+        id=agent_id,
+        subscription_id=body.subscription_id,
+        vm_hash=None,
+        encrypted_secret=encrypted_secret,
+        tags=[agent_id, body.subscription_id, body.account.address],
+    )
+
+    aleph_account = ETHAccount(config.ALEPH_SENDER_SK)
+    async with AuthenticatedAlephHttpClient(
+        account=aleph_account, api_server=config.ALEPH_API_URL
+    ) as client:
+        post_message, _ = await client.create_post(
+            post_content=agent.dict(),
+            post_type=config.ALEPH_AGENT_POST_TYPE,
+            channel=config.ALEPH_CHANNEL,
+        )
+
+
+@app.put("/agent", description="Deploy an agent or update it")
+async def update(body: UpdateAgentPutBody, code: UploadFile, packages: UploadFile):
+    agents = await fetch_agents([body.id])
+
+    if len(agents) != 1:
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail=f"Agent with ID {body.id} not found.",
+        )
+    agent = agents[0]
+    agent_program = (
+        await fetch_agent_program_message(agent.vm_hash)
+        if agent.vm_hash is not None
+        else None
+    )
+
+    decrypted_secret = decrypt(
+        config.ALEPH_SENDER_SK, str.encode(agent.encrypted_secret, "utf-8")
+    ).decode()
+    if body.secret != decrypted_secret:
+        raise HTTPException(
+            status_code=HTTPStatus.UNAUTHORIZED,
+            detail="The secret provided doesn't match the one of this agent.",
+        )
+
+    previous_code_ref = (
+        agent_program.content.code.ref if agent_program is not None else None
+    )
+    # TODO: additional checks on the type of volume, find the right one based on mount etc
+    previous_packages_ref = (
+        agent_program.content.volumes[0].ref if agent_program is not None else None  # type: ignore
+    )
+
+    code_ref = await upload_file(code, previous_code_ref)
+    packages_ref = await upload_file(packages, previous_packages_ref)
+
+    if agent_program is not None:
+        # Program is already deployed and we updated the volumes, exiting here
+        return
+
+    # Register the program
+    aleph_account = ETHAccount(config.ALEPH_SENDER_SK)
+    async with AuthenticatedAlephHttpClient(
+        account=aleph_account, api_server=config.ALEPH_API_URL
+    ) as client:
+        message, _ = await client.create_program(
+            program_ref=code_ref,
+            entrypoint="run",
+            runtime="63f07193e6ee9d207b7d1fcf8286f9aee34e6f12f101d2ec77c1229f92964696",
+            channel=config.ALEPH_CHANNEL,
+            encoding=Encoding.squashfs,
+            persistent=False,
+            volumes=[
+                AlephVolume(
+                    comment="Python packages",
+                    mount="/opt/packages",
+                    ref=packages_ref,
+                    use_latest=True,
+                ).dict()
+            ],
+        )
+
+        # Updating the related POST message
+        await client.create_post(
+            post_content=Agent(
+                **agent.dict(exclude={"vm_hash"}), vm_hash=message.item_hash
+            ),
+            post_type="amend",
+            ref=agent.post_hash,
+            channel=config.ALEPH_CHANNEL,
+        )
+
+
+@app.delete("/agent", description="Remove an agent on subscription end")
+async def delete(body: DeleteAgentBody):
+    # TODO
+    pass

backend/src/utils/agent.py

@@ -0,0 +1,26 @@
+from aleph.sdk import AlephHttpClient
+from aleph.sdk.query.filters import PostFilter
+from aleph_message.models import ProgramMessage
+
+from src.config import config
+from src.interfaces.agent import FetchedAgent
+
+
+async def fetch_agents(ids: list[str] | None = None) -> list[FetchedAgent]:
+    async with AlephHttpClient(api_server=config.ALEPH_API_URL) as client:
+        result = await client.get_posts(
+            post_filter=PostFilter(
+                addresses=[config.ALEPH_SENDER],
+                tags=ids,
+                channels=[config.ALEPH_CHANNEL],
+            )
+        )
+    return [
+        FetchedAgent(**post.content, post_hash=post.item_hash) for post in result.posts
+    ]
+
+
+async def fetch_agent_program_message(item_hash: str) -> ProgramMessage:
+    async with AlephHttpClient(api_server=config.ALEPH_API_URL) as client:
+        result = await client.get_message(item_hash, ProgramMessage)
+        return result

backend/src/utils/storage.py

@@ -0,0 +1,57 @@
+from typing import Any
+
+import aiohttp
+from aleph.sdk import AuthenticatedAlephHttpClient
+from aleph.sdk.chains.ethereum import ETHAccount
+from aleph.sdk.types import StorageEnum
+from aleph_message.models import ItemHash
+from starlette.datastructures import UploadFile
+
+from src.config import config
+
+MAX_DIRECT_STORE_SIZE = 50 * 1024 * 1024  # 50MB
+
+
+async def __upload_on_ipfs(file_content: Any, filename: str | None = None) -> str:
+    """Upload a file on the IPFS gateway of Aleph and return the CID"""
+    async with aiohttp.ClientSession() as session:
+        form_data = aiohttp.FormData()
+        form_data.add_field("file", file_content, filename=filename)
+        response = await session.post(
+            url="https://ipfs.aleph.cloud/api/v0/add", data=form_data
+        )
+        ipfs_data = await response.json()
+        return ipfs_data["Hash"]
+
+
+async def upload_file(file: UploadFile, previous_ref: ItemHash | None = None) -> str:
+    """Upload a file on Aleph, using an IPFS gateway if needed, and returns the STORE message ref"""
+
+    file_content = await file.read()
+    file_size = len(file_content)
+    storage_engine = (
+        StorageEnum.ipfs if file_size > 4 * 1024 * 1024 else StorageEnum.storage
+    )
+
+    aleph_account = ETHAccount(config.ALEPH_SENDER_SK)
+    async with AuthenticatedAlephHttpClient(
+        account=aleph_account, api_server=config.ALEPH_API_URL
+    ) as client:
+        if file_size > MAX_DIRECT_STORE_SIZE:
+            ipfs_hash = await __upload_on_ipfs(file_content, file.filename)
+            store_message, _ = await client.create_store(
+                ref=previous_ref,
+                file_hash=ipfs_hash,
+                storage_engine=storage_engine,
+                channel=config.ALEPH_CHANNEL,
+                guess_mime_type=True,
+            )
+        else:
+            store_message, _ = await client.create_store(
+                ref=previous_ref,
+                file_content=file_content,
+                storage_engine=storage_engine,
+                channel=config.ALEPH_CHANNEL,
+                guess_mime_type=True,
+            )
+        return store_message.item_hash