Skip to content

Runtime Tools

The upjack library provides 6 entity management operations. When using create_server(), these are automatically registered as 6 MCP tools per entity type (e.g., create_contact, get_contact, list_contacts). When using UpjackApp directly, they are available as Python methods.

The operations work on the entity definitions and schemas declared in the app’s upjack manifest. They are the primary interface through which agents (and custom code) interact with entity data.

The current implementation uses in-memory file scanning for search (not a database index). Filters use flat key matching with comparison operators (not JSONPath). Errors are raised as Python exceptions (ValueError, FileNotFoundError, jsonschema.ValidationError). Singleton checking is not yet implemented. Immutable field protection is implemented — update operations silently strip id, type, version, created_at, and created_by from incoming data.

ToolPurposeReturns
create_{entity}Create a new entityCreated entity dict
get_{entity}Get a single entity by IDEntity dict
update_{entity}Update an existing entityUpdated entity dict
list_{plural}Lightweight entity listingArray of entity dicts
search_{plural}Text and structured searchArray of matching entity dicts
delete_{entity}Soft or hard delete an entityDeleted entity dict

When using create_server(), tools are named per entity type:

Entity TypePluralTools Generated
contactcontactscreate_contact, get_contact, update_contact, list_contacts, search_contacts, delete_contact
dealdealscreate_deal, get_deal, update_deal, list_deals, search_deals, delete_deal

Create a new entity of a given type.

app.create_entity(entity_type: str, data: dict, created_by: str = "agent") -> dict
create_{entity}(data: dict) -> dict
ParameterTypeRequiredDefaultDescription
entity_typestringYesEntity type name (e.g., "contact").
dataobjectYesDomain-specific fields for the entity.
created_bystringNo"agent"Origin of the entity (user, agent, system, ingestion, schedule).
  1. Validate entity type: Confirm entity_type exists in the app’s entity definitions. Raise ValueError if not found.
  2. Generate ID: Create a type-prefixed ULID using the entity’s prefix (e.g., ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X).
  3. Inject base fields: Set id, type, version (default 1), created_at (now), updated_at (now), created_by, status ("active").
  4. Extract special fields: Pull tags, relationships, and source from data into base-level fields if present.
  5. Merge data: Combine base fields with the remaining data fields.
  6. Validate against schema: If a schema is loaded for this entity type, validate the complete record. Raise jsonschema.ValidationError on failure — no entity is created.
  7. Write file: Write the entity JSON to {namespace}/data/{plural}/{id}.json with 2-space indentation.
  8. Return: The complete entity dict (base fields + app data merged flat).
from upjack import UpjackApp
app = UpjackApp.from_manifest("manifest.json")
contact = app.create_entity("contact", {
"name": "Alice Chen",
"email": "alice@example.com",
"company_name": "TechCorp",
"title": "VP Engineering",
"stage": "new",
})

Returns:

{
"id": "ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X",
"type": "contact",
"version": 1,
"created_at": "2026-02-15T10:30:00Z",
"updated_at": "2026-02-15T10:30:00Z",
"created_by": "agent",
"status": "active",
"tags": [],
"relationships": [],
"name": "Alice Chen",
"email": "alice@example.com",
"company_name": "TechCorp",
"title": "VP Engineering",
"stage": "new"
}
ConditionExceptionMessage
Unknown entity_typeValueErrorUnknown entity type 'foo'. Known types: ['contact', 'deal']
Schema validation failurejsonschema.ValidationErrorDescribes the specific validation error

Update an existing entity by ID.

app.update_entity(entity_type: str, entity_id: str, data: dict, merge: bool = True) -> dict
update_{entity}(entity_id: str, data: dict) -> dict
ParameterTypeRequiredDefaultDescription
entity_typestringYesEntity type name.
entity_idstringYesEntity ID (e.g., "ct_01HZ...5X").
dataobjectYesFields to update.
mergebooleanNotrueWhether to merge with existing data or replace.
  1. Load entity: Read the entity JSON file from its storage path. Raise FileNotFoundError if not found.
  2. Strip immutable fields: Silently remove id, type, version, created_at, and created_by from incoming data — these cannot be changed after creation.
  3. Merge or replace: If merge: true (default), shallow-merge data into the existing entity. Existing fields not present in data are preserved. If merge: false, replace all fields except immutable base fields.
  4. Auto-update timestamp: Set updated_at to the current timestamp.
  5. Validate against schema: If a schema is loaded, validate the merged entity. Raise jsonschema.ValidationError on failure — no changes are written.
  6. Write file: Overwrite the entity JSON file.
  7. Return: The complete updated entity dict.
updated = app.update_entity("contact", "ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X", {
"stage": "qualified",
"score": 85,
})

Returns:

{
"id": "ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X",
"type": "contact",
"version": 1,
"created_at": "2026-02-15T10:30:00Z",
"updated_at": "2026-02-15T14:22:00Z",
"created_by": "agent",
"status": "active",
"name": "Alice Chen",
"email": "alice@example.com",
"company_name": "TechCorp",
"title": "VP Engineering",
"stage": "qualified",
"score": 85
}
ConditionExceptionMessage
Entity not foundFileNotFoundErrorEntity not found: ct_01HZ...5X
Schema validation failurejsonschema.ValidationErrorDescribes the specific validation error

Text and structured search across entities of a given type.

app.search_entities(
entity_type: str,
query: str | None = None,
filter: dict | None = None,
sort: str = "-updated_at",
limit: int = 20,
) -> list[dict]
search_{plural}(query: str | None, filter: dict | None, sort: str, limit: int) -> list[dict]
ParameterTypeRequiredDefaultDescription
entity_typestringYesEntity type to search.
querystringNoCase-insensitive substring match across all string-valued fields.
filterobjectNoStructured filter with comparison operators.
sortstringNo"-updated_at"Sort field. Prefix with - for descending.
limitintegerNo20Maximum results to return.
  1. Load all entities: Read every JSON file in {namespace}/data/{plural}/. Corrupt files are silently skipped.
  2. Exclude deleted by default: Unless the filter explicitly includes a status key, entities with status: "deleted" are excluded.
  3. Apply text query: If query is provided, match case-insensitively against all string-valued fields in the entity. An entity matches if any string field contains the query as a substring.
  4. Apply structured filters: If filter is provided, evaluate each key-value pair against the entity. All conditions must match (AND logic).
  5. Sort: Apply the sort expression. Default is -updated_at (most recently updated first).
  6. Limit: Return at most limit results.
  7. Return: Array of complete entity dicts matching the query.

Filters use flat field names (not JSONPath) with optional comparison operators:

# Direct equality
app.search_entities("contact", filter={"stage": "qualified"})
# Comparison operators
app.search_entities("contact", filter={"score": {"$gte": 70}})
# Combined filters (AND logic)
app.search_entities("contact", filter={
"stage": {"$in": ["qualified", "contacted"]},
"score": {"$gte": 60},
})

Supported operators:

OperatorMeaningExample
(direct value)Equality{"stage": "qualified"}
$gtGreater than{"score": {"$gt": 50}}
$gteGreater than or equal{"score": {"$gte": 70}}
$ltLess than{"score": {"$lt": 50}}
$lteLess than or equal{"score": {"$lte": 30}}
$neNot equal{"stage": {"$ne": "lost"}}
$inValue in array{"stage": {"$in": ["qualified", "contacted"]}}
$containsArray field contains value{"tags": {"$contains": "saas"}}
$existsField exists (true) or is absent (false){"score": {"$exists": true}}
results = app.search_entities(
"contact",
query="TechCorp",
filter={"score": {"$gte": 60}},
sort="-score",
limit=10,
)

Returns:

[
{
"id": "ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X",
"type": "contact",
"version": 1,
"created_at": "2026-02-15T10:30:00Z",
"updated_at": "2026-02-15T14:22:00Z",
"status": "active",
"name": "Alice Chen",
"email": "alice@example.com",
"company_name": "TechCorp",
"title": "VP Engineering",
"stage": "qualified",
"score": 85
}
]

The v0.1 implementation loads all entity files into memory for each search operation. This works well for typical app sizes (hundreds to low thousands of entities). For larger datasets, a future version may introduce indexing.

Lightweight entity listing. Simpler than entity_search — filters by status only.

app.list_entities(entity_type: str, status: str = "active", limit: int = 50) -> list[dict]
list_{plural}(status: str, limit: int) -> list[dict]
ParameterTypeRequiredDefaultDescription
entity_typestringYesEntity type to list.
statusstringNo"active"Filter by status.
limitintegerNo50Maximum results to return.
  1. Scan directory: List JSON files in {namespace}/data/{plural}/, sorted by updated_at in reverse order (most recently updated first).
  2. Filter by status: Read each entity and include only those matching the status filter. Default is active only.
  3. Limit: Return at most limit results.
  4. Skip corrupt files: Files that fail JSON parsing are silently skipped.
  5. Return: Array of complete entity dicts matching the filter.
contacts = app.list_entities("contact", status="active", limit=5)

Returns:

[
{
"id": "ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X",
"type": "contact",
"status": "active",
"created_at": "2026-02-15T10:30:00Z",
"updated_at": "2026-02-15T14:22:00Z",
"name": "Alice Chen",
"email": "alice@example.com",
"stage": "qualified",
"score": 85
},
{
"id": "ct_01HZ3QLCN8XWVK1QPGB7NU9D6Y",
"type": "contact",
"status": "active",
"created_at": "2026-02-14T09:15:00Z",
"updated_at": "2026-02-14T16:45:00Z",
"name": "Bob Martinez",
"email": "bob@startup.io",
"stage": "contacted",
"score": 62
}
]
Aspectentity_searchentity_list
Text searchYes (substring match)No
Structured filtersYes (comparison operators)Status only
SortAny fieldBy updated_at (most recent first)
Default limit2050
ReturnsFull entity dictsFull entity dicts
Use caseFinding specific entitiesBrowsing/overview

Delete an entity by ID. Soft delete (status change) by default; hard delete (file removal) with explicit flag.

app.delete_entity(entity_type: str, entity_id: str, hard: bool = False) -> dict
delete_{entity}(entity_id: str, hard: bool = False) -> dict
ParameterTypeRequiredDefaultDescription
entity_typestringYesEntity type name.
entity_idstringYesEntity ID to delete.
hardbooleanNofalseIf true, permanently remove the file. If false, set status to "deleted".
  1. Load entity: Read the entity JSON file. Raise FileNotFoundError if not found.
  2. Set status: Change status to "deleted".
  3. Update timestamp: Set updated_at to current time.
  4. Write file: Overwrite the entity JSON file.
  5. Return: The deleted entity dict (with updated status and timestamp).
  1. Load entity: Read the entity JSON file. Raise FileNotFoundError if not found.
  2. Remove file: Delete the JSON file from the workspace.
  3. Return: The entity dict as it was before deletion.
# Soft delete (default)
result = app.delete_entity("contact", "ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X")
assert result["status"] == "deleted"
# Hard delete (permanent)
result = app.delete_entity("contact", "ct_01HZ3QKBN9YWVJ0RPFA7MT8C5X", hard=True)
# File is removed from disk
ConditionExceptionMessage
Entity not foundFileNotFoundErrorEntity not found: ct_01HZ...5X

All operations raise standard Python exceptions:

ExceptionWhen
ValueErrorUnknown entity type, invalid ID prefix
FileNotFoundErrorEntity ID does not exist on disk
jsonschema.ValidationErrorData fails JSON Schema validation
json.JSONDecodeErrorEntity file contains corrupt JSON (raised by get_entity; list and search skip corrupt files silently)

Entity operations work on individual JSON files in the workspace. The consistency model is:

  • Single-writer: The agent processes one tool call at a time. Concurrent writes to the same entity are not expected in v0.1.
  • Read-your-writes: After an operation returns, subsequent reads will see the updated state.
  • No transactions: Multi-entity operations are not atomic. If a skill needs to create a contact and an activity, they are two separate writes. This is acceptable for v0.1 where the agent is the sole writer.

The following features are specified but not yet implemented in v0.1:

  • Singleton constraint: Preventing creation of duplicate singleton entities.
  • Indexed search: SQLite FTS or similar for faster search on large datasets.
  • Structured error objects: MCP-level error responses with code/message/details format.
  • Entity summaries: Returning lightweight projections from entity_list instead of full objects.