5-min Workshop: Tasklist Manager

This workshop builds a complete task tracker. You’ll define entities, route fields across backends, enable semantic search via Vector, attach structured metadata with HyperG’s sub-relational schema, compute derived fields at read time, and expose everything over MCP.

1. Define entities

Every entity has an implicit object_id: Identifier primary key (max 63 chars). Supply one at write time or HeavenBase generates a deterministic hash.

import heavenbase as hb

class Task(hb.Entity):
    title:       hb.ShortText
    description: hb.MediumText
    priority:    hb.Integer
    done:        hb.Boolean
    tags:        hb.Array
    topics:      hb.HyperG

class Developer(hb.Entity):
    name:  hb.ShortText
    email: hb.ShortText

HyperG is HeavenBase’s hypergraph schema type. It creates a sub-relational table owned by each entity row, where you define the schema as triples with customizable columns — {"slot": "...", "value": "...", "topic": "..."}. On SQL backends this becomes a physical side table with one row per triple; on other backends it is stored as a JSON column. Each main entity row can have many HyperG entries, forming a 1:N relationship that you query by conceptually JOINing the main table with this sub-table.

2. Store fields on backends

By default all fields go to main (SQLite). Store Vector on a vector backend for semantic search. Add a compute function so embeddings are derived automatically from the description:

def embed_text(text: str) -> list[float]:
    """Simple embedding: normalize length and 'e' count."""
    return [len(text) / 100.0, float(text.count("e")) / 10.0]


class Task(hb.Entity):
    title:       hb.ShortText
    description: hb.MediumText
    priority:    hb.Integer
    done:        hb.Boolean
    tags:        hb.Array
    topics:      hb.HyperG
    emb:         hb.Vector[2] = hb.field(hb.Vector[2]).store(
                     to="vec", strategy=hb.VectorIndex.name
                 ).compute(embed_text, inputs=["description"])

Fields without explicit storage placement stay on main. The workspace stores emb on vec (in-memory vector backend) and topics in a side table on SQL automatically.

3. Add a computed field

Computed fields are evaluated at read time — they are never stored:

class Developer(hb.Entity):
    name:  hb.ShortText
    email: hb.ShortText
    full:  hb.ShortText = hb.field(hb.ShortText).compute(
        lambda name: f"Dev: {name}",
        inputs=["name"]
    )

Each input field value is passed as a positional argument to the function.

4. Create a workspace and register

ws = hb.HeavenBase("tasklist-demo")
ws.register(Task)
ws.register(Developer)

By default the workspace gets an SQLite backend (main) for structured data and an in-memory vector backend (vec).

5. Write and read tasks

HyperG accepts a list of triples — either dicts or (slot, value, topic) tuples. Each triple becomes one row in the sub-table:

ws.set(Task(id="t1", title="Fix login bug",  description="OAuth token expires after 5 minutes",
             priority=1, done=False, tags=["bug", "auth"],
             topics=[{"slot": "module",   "value": "auth",      "topic": "security"},
                     {"slot": "assigned", "value": "alice",     "topic": "engineering"},
                     {"slot": "sprint",   "value": "sprint-24", "topic": "planning"}]))
ws.set(Task(id="t2", title="Add dark mode",  description="Implement theme toggle component",
             priority=3, done=True,  tags=["feature", "ui"],
             topics=[{"slot": "module",   "value": "ui",        "topic": "frontend"},
                     {"slot": "assigned", "value": "bob",       "topic": "engineering"},
                     {"slot": "sprint",   "value": "sprint-24", "topic": "planning"}]))
ws.set(Task(id="t3", title="Ship v0.1",      description="Tag the release candidate",
             priority=1, done=False, tags=["release"],
             topics=[{"slot": "module",   "value": "release",   "topic": "operations"},
                     {"slot": "assigned", "value": "alice",     "topic": "engineering"},
                     {"slot": "sprint",   "value": "sprint-25", "topic": "planning"}]))
ws.set(Developer(id="d1", name="Alice", email="alice@example.com"))

# Get one by ID — HyperG rows are reconstructed into a list of dicts
task = ws.get("t1", entity=Task)
print(task["title"])       # Fix login bug
print(task["topics"])      # [{'slot': 'module', 'value': 'auth', 'topic': 'security'},
                           #  {'slot': 'assigned', 'value': 'alice', 'topic': 'engineering'},
                           #  {'slot': 'sprint', 'value': 'sprint-24', 'topic': 'planning'}]

# Count
print(ws.count(Task))      # 3

Each HyperG triple becomes a row in the sub-table. The slot column is your custom field key — define any slots your application needs:

`entity_id`	`slot`	`value`	`topic`
t1	module	auth	security
t1	assigned	alice	engineering
t1	sprint	sprint-24	planning
t2	module	ui	frontend
t2	assigned	bob	engineering
t2	sprint	sprint-24	planning

This is a physical 1:N table on SQL — you can think of HyperG as “the entity has a sub-table where each row has slot/value/topic columns that you define.”

6. Batch operations

Insert and retrieve multiple rows at once:

ws.upsert_many(Task, [
    {"id": "t4", "title": "Review PR",      "description": "Check the CI pipeline changes",    "priority": 2, "done": False, "tags": ["devops"],
     "topics": [{"slot": "module", "value": "ci", "topic": "devops"}]},
    {"id": "t5", "title": "Deploy staging", "description": "Push latest build to staging env", "priority": 1, "done": False, "tags": ["release"],
     "topics": [{"slot": "module", "value": "deploy", "topic": "operations"}]},
])

# Get many by ID
tasks = ws.get_many(["t1", "t2", "t5"], entity=Task)
for t in tasks:
    print(t["id"], t["title"])

7. Query with filters

# High-priority incomplete tasks
urgent = list(ws.query(Task).where(Task.priority <= 2).where(Task.done == False))
print(len(urgent))  # 4

# Semantic search via Vector — find tasks related to authentication
near_vec = embed_text("authentication bug")
results = list(ws.query(Task).near(Task.emb, near_vec, top_k=3).select("title", "score"))
for r in results:
    print(r["title"], round(r["score"], 3))
# Fix login bug    0.672
# Deploy staging   0.423
# Review PR        0.318

# Filter by HyperG topic — conceptually a JOIN into the sub-table
security = list(ws.query(Task).where(Task.topics.match("security")))
print(len(security))  # 1 (Fix login bug — its hypergraph has a row where topic='security')

# Filter by HyperG slot value
alice_tasks = list(ws.query(Task).where(Task.topics.match("alice")))
print(len(alice_tasks))  # 2 (t1 and t3 are assigned to alice)

# Get a computed field
dev = ws.get("d1", entity=Developer)
print(dev["full"])  # Dev: Alice

query returns a QueryBuilder. Chain .where() for filters, .near() for vector search, .select() to narrow fields, and .limit() / .offset() for pagination. HyperG filters scan the sub-table and return main-table rows whose sub-table contains a matching entry.

8. Update and delete

# Update — same id, changed fields
ws.set(Task(id="t1", priority=1, done=True))

# Delete — pass entity + id
ws.delete({"entity": Task, "id": "t3"})

print(ws.count(Task))
# 4 (t5 is still there from the batch upsert)

ws.delete() accepts a tuple (entity, entity_id), a dict {"entity": ..., "id": ...}, or just the id when unambiguous.

9. Define from JSON

Entities can also be created from a dict for dynamic use cases — here a Project entity to group tasks:

Project = hb.entity_from_dict({
    "name": "Project",
    "fields": {
        "name":        "ShortText",
        "description": "MediumText",
        "lead":        "ShortText",
    }
})
ws.register(Project)

ws.set(Project(id="p1", name="Sprint 24", description="Q2 delivery", lead="Alice"))

10. Serve as MCP

Expose the workspace as an MCP server in one call:

ws.serve(name="tasklist-mcp")

The server starts on http://127.0.0.1:7001/mcp. Connect any MCP client:

claude mcp add --transport http tasklist-mcp http://127.0.0.1:7001/mcp

Add to opencode.json:

{
  "mcp": {
    "tasklist-mcp": {
      "type": "remote",
      "url": "http://127.0.0.1:7001/mcp"
    }
  }
}

Add to .vscode/mcp.json:

{
  "servers": {
    "tasklist-mcp": {
      "type": "http",
      "url": "http://127.0.0.1:7001/mcp"
    }
  }
}

In Developer > mcp.json:

{
  "mcpServers": {
    "tasklist-mcp": {
      "url": "http://127.0.0.1:7001/mcp",
      "transport": "http"
    }
  }
}

Add to ~/.openclaw/openclaw.json:

{
  "mcpServers": {
    "tasklist-mcp": {
      "url": "http://127.0.0.1:7001/mcp",
      "transport": "http"
    }
  }
}

Add a new MCP server:

Name: tasklist-mcp
Type: Streamable HTTP
URL: http://127.0.0.1:7001/mcp

Add to ~/.hermes/config.yaml:

mcp_servers:
  tasklist-mcp:
    url: "http://127.0.0.1:7001/mcp"

The server runs until interrupted. Use ws.to_mcp_json(name="tasklist-mcp") to print the config without starting the server.

11. Full demo script

Show Complete task tracker

import heavenbase as hb


# ── Helpers ──────────────────────────────────────────────────────────
def embed_text(text: str) -> list[float]:
    return [len(text) / 100.0, float(text.count("e")) / 10.0]


# ── Entities ─────────────────────────────────────────────────────────
class Task(hb.Entity):
    title:       hb.ShortText
    description: hb.MediumText
    priority:    hb.Integer
    done:        hb.Boolean
    tags:        hb.Array
    topics:      hb.HyperG
    emb:         hb.Vector[2] = hb.field(hb.Vector[2]).store(
                     to="vec", strategy=hb.VectorIndex.name
                 ).compute(embed_text, inputs=["description"])


class Developer(hb.Entity):
    name:  hb.ShortText
    email: hb.ShortText
    full:  hb.ShortText = hb.field(hb.ShortText).compute(
        lambda name: f"Dev: {name}", inputs=["name"]
    )


# ── Workspace ────────────────────────────────────────────────────────
ws = hb.HeavenBase("tasklist-demo")
ws.register(Task)
ws.register(Developer)

# ── Seed data ────────────────────────────────────────────────────────
ws.set(Task(id="t1", title="Fix login bug",  description="OAuth token expires after 5 minutes",
             priority=1, done=False, tags=["bug", "auth"],
             topics=[{"slot": "module",   "value": "auth",      "topic": "security"},
                     {"slot": "assigned", "value": "alice",     "topic": "engineering"},
                     {"slot": "sprint",   "value": "sprint-24", "topic": "planning"}]))
ws.set(Task(id="t2", title="Add dark mode",  description="Implement theme toggle component",
             priority=3, done=True,  tags=["feature", "ui"],
             topics=[{"slot": "module",   "value": "ui",        "topic": "frontend"},
                     {"slot": "assigned", "value": "bob",       "topic": "engineering"},
                     {"slot": "sprint",   "value": "sprint-24", "topic": "planning"}]))
ws.set(Task(id="t3", title="Ship v0.1",      description="Tag the release candidate",
             priority=1, done=False, tags=["release"],
             topics=[{"slot": "module",   "value": "release",   "topic": "operations"},
                     {"slot": "assigned", "value": "alice",     "topic": "engineering"},
                     {"slot": "sprint",   "value": "sprint-25", "topic": "planning"}]))
ws.set(Developer(id="d1", name="Alice", email="alice@example.com"))

# ── Batch upsert ─────────────────────────────────────────────────────
ws.upsert_many(Task, [
    {"id": "t4", "title": "Review PR",      "description": "Check the CI pipeline changes",    "priority": 2, "done": False, "tags": ["devops"],
     "topics": [{"slot": "module", "value": "ci", "topic": "devops"}]},
    {"id": "t5", "title": "Deploy staging", "description": "Push latest build to staging env", "priority": 1, "done": False, "tags": ["release"],
     "topics": [{"slot": "module", "value": "deploy", "topic": "operations"}]},
])

# ── Queries ──────────────────────────────────────────────────────────
urgent = list(ws.query(Task).where(Task.priority <= 2).where(Task.done == False))
print(f"Urgent tasks: {len(urgent)}")

near_vec = embed_text("authentication bug")
results = list(ws.query(Task).near(Task.emb, near_vec, top_k=3).select("title", "score"))
for r in results:
    print(f"  {r['title']}: {r['score']:.3f}")

# HyperG filter — scans the sub-table for matching rows
security = list(ws.query(Task).where(Task.topics.match("security")))
print(f"Security tasks: {len(security)}")

alice_tasks = list(ws.query(Task).where(Task.topics.match("alice")))
print(f"Alice's tasks: {len(alice_tasks)}")

dev = ws.get("d1", entity=Developer)
print(f"Developer: {dev['full']}")

# ── Update & delete ──────────────────────────────────────────────────
ws.set(Task(id="t1", priority=1, done=True))
ws.delete({"entity": Task, "id": "t3"})
print(f"Remaining tasks: {ws.count(Task)}")

# ── MCP ──────────────────────────────────────────────────────────────
ws.serve(name="tasklist-mcp")

Further Exploration

Related resources:

HeavenBase MCP — expose an empty workspace, let agents do everything
First MCP — Toolkit persistence and CLI serving
Entities — field types, computed fields, routing
Routing — field-level backend dispatch
Query — filtering, sorting, near, JSON queries
Catalog — object discovery and MetaSchema introspection

Documentation Index

​1. Define entities

​2. Store fields on backends

​3. Add a computed field

​4. Create a workspace and register

​5. Write and read tasks

​6. Batch operations

​7. Query with filters

​8. Update and delete

​9. Define from JSON

​10. Serve as MCP

​11. Full demo script

​Further Exploration

1. Define entities

2. Store fields on backends

3. Add a computed field

4. Create a workspace and register

5. Write and read tasks

6. Batch operations

7. Query with filters

8. Update and delete

9. Define from JSON

10. Serve as MCP

11. Full demo script

Further Exploration