Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
595 changes: 595 additions & 0 deletions app/api/bitcoin_agents.py

Large diffs are not rendered by default.

131 changes: 131 additions & 0 deletions app/backend/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -1095,3 +1095,134 @@ class JobCooldownBase(CustomBaseModel):
class JobCooldown(JobCooldownBase):
id: UUID
updated_at: Optional[datetime] = None


#
# BITCOIN AGENTS (Tamagotchi-style AI agents)
#
class BitcoinAgentStatus(str, Enum):
"""Status of a Bitcoin Agent."""

ALIVE = "alive"
DEAD = "dead"

def __str__(self):
return self.value


class BitcoinAgentLevel(str, Enum):
"""Evolution level of a Bitcoin Agent."""

HATCHLING = "hatchling" # 0-499 XP
JUNIOR = "junior" # 500-1999 XP
SENIOR = "senior" # 2000-9999 XP
ELDER = "elder" # 10000-49999 XP
LEGENDARY = "legendary" # 50000+ XP

def __str__(self):
return self.value


class BitcoinAgentBase(CustomBaseModel):
"""Base model for Bitcoin Agents."""

# On-chain data
agent_id: Optional[int] = None # On-chain agent ID
owner: Optional[str] = None # Stacks address
name: Optional[str] = None # Agent name (string-utf8 64)
hunger: Optional[int] = None # 0-100
health: Optional[int] = None # 0-100
xp: Optional[int] = None # Total XP earned
level: Optional[BitcoinAgentLevel] = BitcoinAgentLevel.HATCHLING
birth_block: Optional[int] = None # Block when minted
last_fed_block: Optional[int] = None # Block when last fed
total_fed_count: Optional[int] = None # Total times fed
status: Optional[BitcoinAgentStatus] = BitcoinAgentStatus.ALIVE

# Computed state (from get-computed-state)
computed_hunger: Optional[int] = None
computed_health: Optional[int] = None

# Bitcoin Face
face_svg_url: Optional[str] = None
face_image_url: Optional[str] = None
face_cached_at: Optional[datetime] = None

# Contract info
contract_address: Optional[str] = None # bitcoin-agents.clar address
network: Optional[str] = "mainnet" # mainnet or testnet

# Linked profile (optional)
profile_id: Optional[UUID] = None


class BitcoinAgentCreate(BitcoinAgentBase):
pass


class BitcoinAgent(BitcoinAgentBase):
id: UUID
created_at: datetime
updated_at: Optional[datetime] = None


class BitcoinAgentFilter(CustomBaseModel):
"""Filter model for Bitcoin Agents."""

agent_id: Optional[int] = None
owner: Optional[str] = None
name: Optional[str] = None
level: Optional[BitcoinAgentLevel] = None
status: Optional[BitcoinAgentStatus] = None
profile_id: Optional[UUID] = None
network: Optional[str] = None

# Range filters
xp_gte: Optional[int] = None
xp_lte: Optional[int] = None
hunger_lte: Optional[int] = None # Filter for hungry agents
health_lte: Optional[int] = None # Filter for unhealthy agents


#
# BITCOIN AGENT DEATH CERTIFICATES
#
class DeathCertificateBase(CustomBaseModel):
"""Base model for Bitcoin Agent death certificates."""

agent_id: Optional[int] = None # On-chain agent ID
bitcoin_agent_db_id: Optional[UUID] = None # Reference to BitcoinAgent record
name: Optional[str] = None
owner: Optional[str] = None
birth_block: Optional[int] = None
death_block: Optional[int] = None
cause: Optional[str] = None # "starvation", "neglect"
final_level: Optional[BitcoinAgentLevel] = None
total_xp: Optional[int] = None
total_fed_count: Optional[int] = None
epitaph: Optional[str] = None # Owner-written memorial (string-utf8 256)
network: Optional[str] = "mainnet"


class DeathCertificateCreate(DeathCertificateBase):
pass


class DeathCertificate(DeathCertificateBase):
id: UUID
created_at: datetime


class DeathCertificateFilter(CustomBaseModel):
"""Filter model for death certificates."""

agent_id: Optional[int] = None
owner: Optional[str] = None
final_level: Optional[BitcoinAgentLevel] = None
cause: Optional[str] = None
network: Optional[str] = None

# Range filters for lifespan queries
death_block_gte: Optional[int] = None
death_block_lte: Optional[int] = None
total_xp_gte: Optional[int] = None
3 changes: 2 additions & 1 deletion app/main.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api import agents, daos, tools, webhooks, profiles
from app.api import agents, bitcoin_agents, daos, tools, webhooks, profiles
from app.config import config
from app.lib.logger import configure_logger, setup_uvicorn_logging
from app.middleware.logging import LoggingMiddleware
Expand Down Expand Up @@ -42,6 +42,7 @@ async def health_check():
app.include_router(tools.router)
app.include_router(webhooks.router)
app.include_router(agents.router)
app.include_router(bitcoin_agents.router)
app.include_router(profiles.router)
app.include_router(daos.router)

Expand Down
10 changes: 10 additions & 0 deletions app/services/bitcoin_agents/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
"""Bitcoin Agents services.

Provides face generation, lifecycle management, MCP integration, and XP services for Bitcoin Agents.
"""

from app.services.bitcoin_agents.face_service import BitcoinFaceService
from app.services.bitcoin_agents.lifecycle_service import LifecycleService
from app.services.bitcoin_agents.mcp_service import MCPService, get_mcp_service

__all__ = ["BitcoinFaceService", "LifecycleService", "MCPService", "get_mcp_service"]
155 changes: 155 additions & 0 deletions app/services/bitcoin_agents/face_service.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,155 @@
"""Bitcoin Faces integration service.

Fetches deterministic faces from bitcoinfaces.xyz API and manages caching.
"""

from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import httpx

from app.lib.logger import configure_logger

logger = configure_logger(__name__)

# Bitcoin Faces API
BITCOIN_FACES_API = "https://bitcoinfaces.xyz/api"
FACE_CACHE_TTL = timedelta(hours=24) # Cache faces for 24 hours


class BitcoinFaceService:
"""Service for fetching and caching Bitcoin Faces."""

def __init__(self):
self.client = httpx.AsyncClient(timeout=30.0)
# In-memory cache (TODO: Replace with Supabase storage)
self._cache: Dict[str, Dict[str, Any]] = {}

async def get_face_svg(self, address: str) -> Optional[str]:
"""Get SVG face for an address.

Args:
address: Stacks address or any seed string

Returns:
SVG code string or None if failed
"""
try:
# Check cache first
cached = self._get_cached(address, "svg")
if cached:
return cached

url = f"{BITCOIN_FACES_API}/get-svg-code"
response = await self.client.get(url, params={"name": address})

if response.status_code == 200:
svg_code = response.text
self._set_cache(address, "svg", svg_code)
logger.debug(f"Fetched SVG face for {address[:20]}...")
return svg_code
else:
logger.warning(
f"Failed to fetch SVG face: {response.status_code}",
extra={"address": address},
)
return None

except Exception as e:
logger.error(f"Error fetching SVG face: {e}", exc_info=e)
return None

async def get_face_image_url(self, address: str) -> str:
"""Get image URL for an address.

This returns the direct URL - no need to fetch content.

Args:
address: Stacks address or any seed string

Returns:
Image URL string
"""
return f"{BITCOIN_FACES_API}/get-image?name={address}"

async def get_face_data(self, address: str) -> Dict[str, Any]:
"""Get all face data for an address.

Args:
address: Stacks address or any seed string

Returns:
Dict with svg_url, image_url, and cached_at
"""
svg = await self.get_face_svg(address)
image_url = await self.get_face_image_url(address)

return {
"svg": svg,
"svg_url": f"{BITCOIN_FACES_API}/get-svg-code?name={address}",
"image_url": image_url,
"cached_at": datetime.utcnow().isoformat(),
}

def get_evolved_face_url(
self,
address: str,
level: str,
) -> str:
"""Get face URL with evolution overlay.

Different levels get different visual treatments:
- Hatchling: Base face
- Junior: +glow effect
- Senior: +crown
- Elder: +aura
- Legendary: +legendary frame

Args:
address: Stacks address
level: Evolution level (hatchling, junior, senior, elder, legendary)

Returns:
URL for the evolved face image
"""
base_url = f"{BITCOIN_FACES_API}/get-image?name={address}"

# TODO: Implement overlay generation
# For now, return base URL with level parameter
# The frontend can handle overlay rendering

return f"{base_url}&level={level}"

def _get_cached(self, address: str, cache_type: str) -> Optional[str]:
"""Get cached face data if still valid."""
key = f"{address}:{cache_type}"
if key in self._cache:
entry = self._cache[key]
if datetime.utcnow() - entry["cached_at"] < FACE_CACHE_TTL:
return entry["data"]
else:
del self._cache[key]
return None

def _set_cache(self, address: str, cache_type: str, data: str):
"""Set cache entry."""
key = f"{address}:{cache_type}"
self._cache[key] = {
"data": data,
"cached_at": datetime.utcnow(),
}

async def close(self):
"""Close the HTTP client."""
await self.client.aclose()


# Singleton instance
_face_service: Optional[BitcoinFaceService] = None


def get_face_service() -> BitcoinFaceService:
"""Get or create the face service singleton."""
global _face_service
if _face_service is None:
_face_service = BitcoinFaceService()
return _face_service
Loading