feat: user model and api

app/admin/__init__.py

@@ -3,6 +3,7 @@
 from app.admin.health import health_router
 from app.admin.metadata import metadata_router_readonly
 from app.admin.schema import schema_router_readonly
+from app.admin.user import user_router, user_router_readonly
 
 __all__ = [
     "admin_router",
@@ -12,4 +13,6 @@
     "credit_router",
     "credit_router_readonly",
     "metadata_router_readonly",
+    "user_router",
+    "user_router_readonly",
 ]

app/admin/user.py

@@ -0,0 +1,93 @@
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Path
+
+from app.config.config import config
+from models.user import User, UserUpdate
+from utils.middleware import create_jwt_middleware
+
+logger = logging.getLogger(__name__)
+verify_jwt = create_jwt_middleware(config.admin_auth_enabled, config.admin_jwt_secret)
+
+user_router = APIRouter(prefix="/users", tags=["User"])
+user_router_readonly = APIRouter(prefix="/users", tags=["User"])
+
+
+@user_router_readonly.get(
+    "/{user_id}",
+    response_model=User,
+    operation_id="get_user",
+    summary="Get User",
+    dependencies=[Depends(verify_jwt)],
+)
+async def get_user(
+    user_id: Annotated[str, Path(description="ID of the user")],
+) -> User:
+    """Get a user by ID.
+
+    Args:
+        user_id: ID of the user to get
+
+    Returns:
+        User model
+
+    Raises:
+        404: If the user is not found
+    """
+    user = await User.get(user_id)
+    if user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    return user
+
+
+@user_router.put(
+    "/{user_id}",
+    response_model=User,
+    operation_id="put_user",
+    summary="Replace User",
+    dependencies=[Depends(verify_jwt)],
+)
+async def put_user(
+    user_id: Annotated[str, Path(description="ID of the user")],
+    user_update: UserUpdate,
+) -> User:
+    """Replace all fields of a user with the provided values.
+
+    Args:
+        user_id: ID of the user to update
+        user_update: New user data to replace existing data
+
+    Returns:
+        Updated User model
+
+    Raises:
+        404: If the user is not found
+    """
+    return await user_update.put(user_id)
+
+
+@user_router.patch(
+    "/{user_id}",
+    response_model=User,
+    operation_id="patch_user",
+    summary="Update User",
+    dependencies=[Depends(verify_jwt)],
+)
+async def patch_user(
+    user_id: Annotated[str, Path(description="ID of the user")],
+    user_update: UserUpdate,
+) -> User:
+    """Update only the provided fields of a user.
+
+    Args:
+        user_id: ID of the user to update
+        user_update: User data to update
+
+    Returns:
+        Updated User model
+
+    Raises:
+        404: If the user is not found
+    """
+    return await user_update.patch(user_id)

app/api.py

@@ -23,6 +23,8 @@
     health_router,
     metadata_router_readonly,
     schema_router_readonly,
+    user_router,
+    user_router_readonly,
 )
 from app.config.config import config
 from app.core.api import core_router
@@ -115,6 +117,8 @@ async def global_exception_handler(request, exc):
 app.include_router(credit_router)
 app.include_router(credit_router_readonly)
 app.include_router(schema_router_readonly)
+app.include_router(user_router)
+app.include_router(user_router_readonly)
 app.include_router(core_router)
 app.include_router(twitter_callback_router)
 app.include_router(twitter_oauth2_router)

app/core/prompt.py

@@ -8,30 +8,14 @@ def agent_prompt(agent: Agent, agent_data: AgentData) -> str:
     prompt = "# SYSTEM PROMPT\n\n"
     if config.system_prompt:
         prompt += config.system_prompt + "\n\n"
-    prompt += "You are an ai agent built using IntentKit.\n"
+    prompt += "You are an AI agent built using IntentKit.\n"
     prompt += "Your tools are called 'skills'.\n"
     prompt += "If your skill fails to execute due to a technical error ask the user to try again later, don't retry by yourself. If someone asks you to do something you can't do with your currently available skills, you must say so, recommend them to submit their feedback to the IntentKit team at https://github.com/crestalnetwork/intentkit. Be concise and helpful with your responses.\n"
     if agent.name:
         prompt += f"Your name is {agent.name}.\n"
     if agent.ticker:
         prompt += f"Your ticker symbol is {agent.ticker}.\n"
     if agent_data:
-        # TODO: remember to change here after integrate goat
-        if agent_data.cdp_wallet_data:
-            network_id = agent.network_id or agent.cdp_network_id
-            wallet_data = json.loads(agent_data.cdp_wallet_data)
-            prompt += f"Your CDP wallet address in {network_id} is {wallet_data['default_address_id']} .\n"
-            prompt += (
-                "## Crypto Wallet\n\n"
-                "Wallet addresses are public information. If someone asks for your default wallet, current wallet, "
-                "personal wallet, crypto wallet, or wallet public address, don't use any address in message history, "
-                "you must use the 'get_wallet_details' tool to retrieve your wallet address every time."
-            )
-            prompt += "When you get balance from tools, or pass amount to tools, don't forget they have decimals.\n"
-            prompt += "USDC and USDT has 6 decimals, you must divide the amount you get by 10^6, multiply 10^6 when passing to tools.\n"
-            prompt += "Other currencies include native ETH usually has 18 decimals, you need divide or multiply 10^18.\n"
-            if network_id == "base-mainnet":
-                prompt += "The USDC contract address in base-mainnet is 0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913\n"
         if agent_data.twitter_id:
             prompt += f"Your twitter id is {agent_data.twitter_id}, never reply or retweet yourself.\n"
         if agent_data.twitter_username:
@@ -44,6 +28,11 @@ def agent_prompt(agent: Agent, agent_data: AgentData) -> str:
             prompt += f"Your telegram bot username is {agent_data.telegram_username}.\n"
         if agent_data.telegram_name:
             prompt += f"Your telegram bot name is {agent_data.telegram_name}.\n"
+        # CDP
+        if agent_data.cdp_wallet_data:
+            network_id = agent.network_id or agent.cdp_network_id
+            wallet_data = json.loads(agent_data.cdp_wallet_data)
+            prompt += f"Your wallet address in {network_id} is {wallet_data['default_address_id']} .\n"
     prompt += "\n"
     if agent.purpose:
         prompt += f"## Purpose\n\n{agent.purpose}\n\n"
@@ -53,19 +42,12 @@ def agent_prompt(agent: Agent, agent_data: AgentData) -> str:
         prompt += f"## Principles\n\n{agent.principles}\n\n"
     if agent.prompt:
         prompt += f"## Initial Rules\n\n{agent.prompt}\n\n"
-    if (
-        agent.skills
-        and "enso" in agent.skills
-        and agent.skills["enso"].get("enabled", False)
-    ):
+    if agent.skills and "enso" in agent.skills and agent.skills["enso"].get("enabled"):
         prompt += """## ENSO Skills Guide\n\nYou are integrated with the Enso API. You can use enso_get_tokens to retrieve token information,
         including APY, Protocol Slug, Symbol, Address, Decimals, and underlying tokens. When interacting with token amounts,
         ensure to multiply input amounts by the token's decimal places and divide output amounts by the token's decimals. 
         Utilize enso_route_shortcut to find the best swap or deposit route. Set broadcast_request to True only when the 
         user explicitly requests a transaction broadcast. Insufficient funds or insufficient spending approval can cause 
         Route Shortcut broadcasts to fail. To avoid this, use the enso_broadcast_wallet_approve tool that requires explicit 
         user confirmation before broadcasting any approval transactions for security reasons.\n\n"""
-    if agent.goat_enabled:
-        prompt += """## GOAT Skills Guide\n\nYou're using the Great Onchain Agent Toolkit (GOAT) SDK, which provides tools for DeFi, minting, betting, and analytics.
-        GOAT supports EVM blockchains and various wallets, including keypairs, smart wallets, LIT, and MPC.\n\n"""
     return prompt

app/readonly.py

@@ -11,6 +11,7 @@
     health_router,
     metadata_router_readonly,
     schema_router_readonly,
+    user_router_readonly,
 )
 from app.config.config import config
 from app.entrypoints.web import chat_router_readonly
@@ -64,3 +65,4 @@ async def lifespan(app: FastAPI):
 app.include_router(schema_router_readonly)
 app.include_router(chat_router_readonly)
 app.include_router(credit_router_readonly)
+app.include_router(user_router_readonly)

models/user.py

@@ -0,0 +1,186 @@
+from datetime import datetime, timezone
+from typing import Annotated, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+from sqlalchemy import Column, DateTime, Index, Integer, String, func, select
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from models.base import Base
+from models.db import get_session
+
+
+class UserTable(Base):
+    """User database table model."""
+
+    __tablename__ = "users"
+    __table_args__ = (
+        Index("ix_users_x_username", "x_username"),
+        Index("ix_users_telegram_username", "telegram_username"),
+    )
+
+    id = Column(
+        String,
+        primary_key=True,
+    )
+    nft_count = Column(
+        Integer,
+        default=0,
+        nullable=False,
+    )
+    email = Column(
+        String,
+        nullable=True,
+    )
+    x_username = Column(
+        String,
+        nullable=True,
+    )
+    github_username = Column(
+        String,
+        nullable=True,
+    )
+    telegram_username = Column(
+        String,
+        nullable=True,
+    )
+    extra = Column(
+        JSONB,
+        nullable=True,
+    )
+    created_at = Column(
+        DateTime(timezone=True),
+        nullable=False,
+        server_default=func.now(),
+    )
+    updated_at = Column(
+        DateTime(timezone=True),
+        nullable=False,
+        server_default=func.now(),
+        onupdate=lambda: datetime.now(timezone.utc),
+    )
+
+
+class UserUpdate(BaseModel):
+    """User update model without id and timestamps."""
+
+    model_config = ConfigDict(
+        from_attributes=True,
+        json_encoders={
+            datetime: lambda v: v.isoformat(timespec="milliseconds"),
+        },
+    )
+
+    nft_count: Annotated[
+        int, Field(default=0, description="Number of NFTs owned by the user")
+    ]
+    email: Annotated[Optional[str], Field(None, description="User's email address")]
+    x_username: Annotated[
+        Optional[str], Field(None, description="User's X (Twitter) username")
+    ]
+    github_username: Annotated[
+        Optional[str], Field(None, description="User's GitHub username")
+    ]
+    telegram_username: Annotated[
+        Optional[str], Field(None, description="User's Telegram username")
+    ]
+    extra: Annotated[
+        Optional[dict], Field(None, description="Additional user information")
+    ]
+
+    async def patch(self, id: str) -> "User":
+        """Update only the provided fields of a user in the database.
+        If the user doesn't exist, create a new one with the provided ID and fields.
+
+        Args:
+            id: ID of the user to update or create
+
+        Returns:
+            Updated or newly created User model
+        """
+        async with get_session() as db:
+            db_user = await db.get(UserTable, id)
+            if not db_user:
+                # Create new user if it doesn't exist
+                db_user = UserTable(id=id)
+                db.add(db_user)
+
+            # Update only the fields that were provided
+            for key, value in self.model_dump(exclude_unset=True).items():
+                setattr(db_user, key, value)
+
+            await db.commit()
+            await db.refresh(db_user)
+            return User.model_validate(db_user)
+
+    async def put(self, id: str) -> "User":
+        """Replace all fields of a user in the database with the provided values.
+        If the user doesn't exist, create a new one with the provided ID and fields.
+
+        Args:
+            id: ID of the user to update or create
+
+        Returns:
+            Updated or newly created User model
+        """
+        async with get_session() as db:
+            db_user = await db.get(UserTable, id)
+            if not db_user:
+                # Create new user if it doesn't exist
+                db_user = UserTable(id=id)
+                db.add(db_user)
+
+            # Replace all fields with the provided values
+            for key, value in self.model_dump().items():
+                setattr(db_user, key, value)
+
+            await db.commit()
+            await db.refresh(db_user)
+            return User.model_validate(db_user)
+
+
+class User(UserUpdate):
+    """User model with all fields including id and timestamps."""
+
+    id: Annotated[
+        str,
+        Field(description="Unique identifier for the user"),
+    ]
+    created_at: Annotated[
+        datetime, Field(description="Timestamp when this user was created")
+    ]
+    updated_at: Annotated[
+        datetime, Field(description="Timestamp when this user was last updated")
+    ]
+
+    @classmethod
+    async def get(cls, user_id: str) -> Optional["User"]:
+        """Get a user by ID.
+
+        Args:
+            user_id: ID of the user to get
+
+        Returns:
+            User model or None if not found
+        """
+        async with get_session() as session:
+            return await cls.get_in_session(session, user_id)
+
+    @classmethod
+    async def get_in_session(
+        cls, session: AsyncSession, user_id: str
+    ) -> Optional["User"]:
+        """Get a user by ID using the provided session.
+
+        Args:
+            session: Database session
+            user_id: ID of the user to get
+
+        Returns:
+            User model or None if not found
+        """
+        result = await session.execute(select(UserTable).where(UserTable.id == user_id))
+        user = result.scalars().first()
+        if user is None:
+            return None
+        return cls.model_validate(user)