When two agents try to update conversation history at the same time, one update can overwrite the other. When an AI workflow crashes mid-execution, the conversation state is lost. Normal in-memory state doesn’t survive failures or coordinate concurrent access.

AGNT5 entities solve this problem. Each entity instance is identified by a unique key with automatic state persistence and single-writer consistency. State survives restarts, concurrent updates are serialized per key, and you write regular Python classes.

Example

from agnt5 import Entity

class Conversation(Entity):
    async def add_message(self, role: str, content: str) -> dict:
        messages = self.state.get("messages", [])
        messages.append({"role": role, "content": content})
        self.state.set("messages", messages)
        return {"message_count": len(messages)}

    async def get_history(self) -> list[dict]:
        return self.state.get("messages", [])

When you call it:

from agnt5 import Client

client = Client()

# Each key is an isolated entity instance
conversation = client.entity("Conversation", "session-abc")
conversation.add_message(role="user", content="Explain quantum computing")

# Same key = same state
history = conversation.get_history()  # Returns all messages

AGNT5:

• Creates an isolated instance per key (one conversation per session)
• Loads state before each method call
• Executes the method with state accessible via self.state
• Saves state automatically after successful execution
• Guarantees single-writer consistency (no concurrent modifications per key)

If the process crashes after add_message but before get_history, the state persists. When you call get_history, AGNT5 loads the saved state — no data loss.

Entity State

Access state through the self.state interface inside entity methods:

from agnt5 import Entity

class ResearchSession(Entity):
    async def add_finding(self, source: str, content: str) -> dict:
        # Get current findings
        findings = self.state.get("findings", [])

        # Add new finding
        findings.append({"source": source, "content": content})
        self.state.set("findings", findings)
        self.state.set("last_updated", datetime.now().isoformat())

        return {
            "total_findings": len(findings),
            "latest_source": source
        }

    async def clear_findings(self) -> dict:
        # Delete specific key
        self.state.delete("findings")
        return {"findings_cleared": True}

    async def reset_session(self) -> dict:
        # Clear all state
        self.state.clear()
        return {"session_reset": True}

Available operations:

• self.state.get(key, default) — Get value from state (returns default if not found)
• self.state.set(key, value) — Set value in state
• self.state.delete(key) — Delete key from state
• self.state.clear() — Clear all state

Important: State is only accessible inside entity methods. You cannot access conversation.state directly outside a method — state is loaded/saved automatically around method execution.

Single-Writer Consistency

Each entity key has single-writer consistency: only one method call can modify state at a time for a given key.

from agnt5 import Entity

class AgentMemory(Entity):
    async def store_fact(self, fact: str, importance: float) -> dict:
        facts = self.state.get("facts", [])
        facts.append({"content": fact, "importance": importance})
        self.state.set("facts", facts)
        return {"total_facts": len(facts)}

When you call this concurrently:

# Two concurrent updates on the same key
memory = client.entity("AgentMemory", "agent-001")
result1 = memory.store_fact("User prefers technical details", 0.9)  # First call
result2 = memory.store_fact("User is interested in AI", 0.8)  # Second call (waits for first)

AGNT5 serializes these calls:

1. Load state (facts = [])
2. Execute first store_fact, save state (facts = [fact1])
3. Load state (facts = [fact1])
4. Execute second store_fact, save state (facts = [fact1, fact2])

Different keys run in parallel:

# Different keys = different instances = parallel execution
memory_a = client.entity("AgentMemory", "agent-001")
memory_b = client.entity("AgentMemory", "agent-002")

# These run concurrently (different agents)
result_a = memory_a.store_fact("Fact for agent 1", 0.9)
result_b = memory_b.store_fact("Fact for agent 2", 0.8)

This is how entities provide isolation: each key is an independent instance with its own state and serialized access.

Entity Instances by Key

Entities are keyed instances. The key identifies which instance you’re operating on.

from agnt5 import Entity

class AgentContext(Entity):
    async def add_context(self, context_type: str, data: dict) -> dict:
        contexts = self.state.get("contexts", {})
        contexts[context_type] = data
        self.state.set("contexts", contexts)
        return {"total_contexts": len(contexts)}

    async def get_context(self, context_type: str) -> dict:
        contexts = self.state.get("contexts", {})
        return contexts.get(context_type, {})

Each key is a separate agent context:

# Research agent's context
research_agent = client.entity("AgentContext", "research-agent-001")
research_agent.add_context("domain", {"field": "quantum computing"})
domain_context = research_agent.get_context("domain")  # {"field": "quantum computing"}

# Analysis agent's context (completely separate state)
analysis_agent = client.entity("AgentContext", "analysis-agent-001")
analysis_agent.add_context("domain", {"field": "financial markets"})
domain_context = analysis_agent.get_context("domain")  # {"field": "financial markets"}

Common key patterns:

• Agent-scoped: AgentMemory with key agent-001
• Session-scoped: Conversation with key session-abc
• User-scoped: UserPreferences with key user-123
• Task-scoped: ResearchTask with key task-456

Choose keys based on your isolation boundary: what should share state and what should be independent.

Calling Entities

From Client SDK

from agnt5 import Client

client = Client()

# Get entity instance
conversation = client.entity("Conversation", "session-abc")

# Call methods
conversation.add_message(role="user", content="Explain quantum computing")
history = conversation.get_history()

The client handles serialization, routing to the correct worker, and state management.

State Persistence

Entity state is automatically persisted after each method execution:

1. Load state — Before method runs, current state is loaded for the key
2. Execute method — Your code runs with self.state access
3. Save state — After successful execution, state is persisted
4. Optimistic locking — Version check ensures no concurrent modifications

If a method throws an exception, state changes are not saved. This provides transactional semantics: either the method succeeds and state is saved, or it fails and state remains unchanged.

class TokenBudget(Entity):
    async def consume_tokens(self, tokens: int) -> dict:
        budget = self.state.get("remaining_tokens", 0)

        if tokens > budget:
            # State NOT saved (exception thrown)
            raise ValueError(f"Insufficient token budget: {budget} < {tokens}")

        # Only saved if we get here successfully
        budget -= tokens
        self.state.set("remaining_tokens", budget)
        return {"remaining_tokens": budget}

When to Use Entities

Use entities when you need:

• Stateful objects — Conversation history, agent memory, user preferences, research sessions
• Isolation by key — Each agent/session/user needs independent state
• Coordinated updates — Prevent concurrent modification conflicts
• Object-oriented APIs — Clean class-based abstractions with methods

Use functions instead when:

• Computation is purely stateless (LLM calls, embeddings, analysis)
• You don’t need state to persist across calls
• No coordination between concurrent calls is needed

Entities as Foundation

Entities provide stateful building blocks for AI applications:

• Functions compute outputs from inputs (stateless LLM calls, analysis, embeddings)
• Entities manage state with single-writer consistency (conversation history, agent memory)
• Higher-level patterns build on both (multi-step AI workflows, agent coordination)

What makes entities reliable is how AGNT5 treats each key: as an isolated instance with automatic persistence and serialized access. When you update state, it survives crashes. When multiple calls happen concurrently, they execute one at a time.

This is different from normal objects in two ways: state persists beyond process lifetime, and concurrent access is automatically coordinated. You write regular Python classes, and AGNT5 handles the distributed systems complexity — single-writer guarantees, optimistic concurrency, durable storage, and crash recovery.

When a conversation update happens mid-crash, when two agents access the same memory, when AI workflows coordinate shared state — entities ensure data integrity without manual locking or state management code. That’s what makes AI applications reliable.