Thread-Safe Agent State

Thread-safe chat history and cache management

PraisonAI protects chat history, caches, and session state when agents run from multiple threads or async tasks.

from praisonaiagents import Agent
import threading

agent = Agent(name="Assistant", instructions="You are helpful.")

def worker(prompt):
    agent.chat(prompt)

threads = [threading.Thread(target=worker, args=(f"Question {i}",)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

The user drives one agent from several threads; locks keep chat history and caches consistent without corrupted state.

How It Works

Quick Start

1. Multi-threaded Chat

   from praisonaiagents import Agent
   import threading
   
   agent = Agent(name="ThreadSafeAgent", instructions="You are helpful.")
   
   def worker(prompt):
       response = agent.chat(prompt)
       print(f"Response: {response[:50]}...")
   
   threads = [threading.Thread(target=worker, args=(f"Question {i}",)) for i in range(5)]
   for t in threads:
       t.start()
   for t in threads:
       t.join()
   
2. Async Concurrent Tasks

   import asyncio
   from praisonaiagents import Agent
   
   agent = Agent(name="AsyncAgent")
   
   async def async_chat(prompt):
       return agent.chat(prompt)
   
   async def main():
       results = await asyncio.gather(*[async_chat(f"Question {i}") for i in range(5)])
   
   asyncio.run(main())
   

Behaviour change in PR #1548: run_sync() now raises RuntimeError when called from inside a running event loop. Previously it would auto-fallback to a background loop.

# Before PR #1548 (worked, but unsafe)
async def handler():
    result = run_sync(some_coro())  # silently used background loop

# After PR #1548 (raises RuntimeError)
async def handler():
    result = await some_coro()  # use this from async context

Thread-Safe Components

Chat History

The chat_history property is now fully thread-safe with automatic locking. The SDK protects chat history mutations through internal helper methods and a locked setter:

from praisonaiagents import Agent
import threading

agent = Agent(
    name="ThreadSafeAgent",
    instructions="You are helpful."
)

def worker(prompt):
    # Safe to call from multiple threads
    response = agent.chat(prompt)
    print(f"Response: {response[:50]}...")

# Create multiple threads
threads = [
    threading.Thread(target=worker, args=(f"Question {i}",))
    for i in range(5)
]

# Start all threads
for t in threads:
    t.start()

# Wait for completion
for t in threads:
    t.join()

What changed in PR #1488

Prior to PR #1488, chat_history mutations bypassed thread-safety locks at 31+ call sites. The SDK now uses internal helper methods that properly acquire locks:

• _append_to_chat_history(message) - Thread-safe message appending
• _truncate_chat_history(length) - Thread-safe history truncation
• _replace_chat_history(new_history) - Thread-safe full replacement
• chat_history setter now acquires the AsyncSafeState lock for assignments

What changed in PR #1514

PR #1514 enhanced thread-safety in three key areas:

1. Locked Memory Initialization: Task.initialize_memory() now uses threading.Lock with double-checked locking pattern. A new async variant initialize_memory_async() uses asyncio.Lock and offloads construction with asyncio.to_thread() to prevent event loop blocking.

2. Async-Locked Workflow State: New _set_workflow_finished(value) method uses async locks to safely update workflow completion status across concurrent tasks.

3. Non-Mutating Task Context: Task execution no longer mutates task.description during runs. Per-execution context is stored in _execution_context field, keeping the user-facing task.description stable across multiple executions.

Safe operations

# These operations are now thread-safe out of the box:
agent.chat_history = []  # Full replacement - uses locked setter
agent.chat("Hello")      # Appends safely via internal methods

# Reading is always safe:
history = agent.chat_history
print(f"History has {len(history)} messages")

Caches

Internal caches use threading.RLock for reentrant locking:

• _system_prompt_cache - Cached system prompts
• _formatted_tools_cache - Cached tool definitions

Rate Limiter

RateLimiter can be shared across threads and agents. Both the sync and async method families are fully locked — see Rate Limiter → Thread Safety & Multi-Agent Use for patterns.

LiteAgent Thread Safety

The lite package also provides thread-safe operations:

from praisonaiagents.lite import LiteAgent, create_openai_llm_fn
import threading

llm_fn = create_openai_llm_fn(model="gpt-4o-mini")
agent = LiteAgent(name="LiteThreadSafe", llm_fn=llm_fn)

def concurrent_chat(message):
    return agent.chat(message)

# Safe concurrent access
with threading.ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(concurrent_chat, f"Q{i}") for i in range(10)]
    results = [f.result() for f in futures]

Implementation Details

Lock Types

Component	Lock Type	Reason
chat_history	AsyncSafeState (DualLock: RLock + per-loop asyncio.Lock)	Re-entrant on same thread; non-blocking in async contexts
_cache_lock	threading.RLock	Allows reentrant access from cached helpers
PersistenceOrchestrator._cache_lock	threading.RLock	Protects _session_cache reads/writes; reads return deepcopy() to prevent shared mutable state
RateLimiter (sync)	threading.Lock	Protects _tokens, _api_tokens, and refill state from races in multi-threaded acquire calls
RateLimiter (async)	asyncio.Lock	Same protection for coroutine contexts

Deep-copy support

Agent.deepcopy replaces threading.RLock (__cache_lock) and threading.Lock (_cost_lock) with fresh instances during copy.deepcopy(agent), so the agent is copy-safe on every supported Python version (RLock pickling was broken on CPython < 3.13).

Lock Usage Pattern

# Internal implementation pattern
class Agent:
    def __init__(self):
        self.__chat_history_state = AsyncSafeState([])
        self._cache_lock = threading.RLock()
    
    @property
    def _history_lock(self):
        return self.__chat_history_state
    
    def _append_to_chat_history(self, message):
        with self._history_lock.lock():
            self._history_lock.value.append(message)

Why a re-entrant lock?

Nested calls (e.g. a helper that holds the lock and then assigns chat_history, which itself acquires the lock) used to deadlock. RLock permits the same thread to re-enter. See PR #1567 for details.

Persistence Orchestrator session cache

PersistenceOrchestrator now guards its in-memory session cache with threading.RLock and returns deep copies on read, so concurrent agents can share an orchestrator without corrupting cached ConversationSession objects.

from praisonaiagents import Agent
from praisonai.persistence import PersistenceOrchestrator
import threading

# Create shared orchestrator
orchestrator = PersistenceOrchestrator()

def agent_worker(agent_name, session_id):
    agent = Agent(name=agent_name)
    
    # Safe: orchestrator returns deep copy of cached session
    history = orchestrator.on_agent_start(agent, session_id=session_id)
    
    # Multiple agents can safely read/update sessions concurrently
    orchestrator.on_message(session_id, "user", "Hello from " + agent_name)
    orchestrator.on_agent_end(agent, session_id)

# Multiple threads can safely share the orchestrator
threads = [
    threading.Thread(target=agent_worker, args=(f"Agent{i}", "session-123"))
    for i in range(3)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

Reference: PraisonAI PR #1609. The session cache uses defensive copying to prevent shared mutable state between concurrent operations.

DefaultSessionStore — cross-process safety

DefaultSessionStore uses a per-session file lock (fcntl.flock on Unix, msvcrt.locking on Windows) for every write path:

Method	Locked read-modify-write	Atomic write
add_message()	✅	✅
add_user_message() / add_assistant_message()	✅	✅
clear_session()	✅	✅
update_session_metadata()	✅ (since PR #1709)	✅
set_agent_info()	in-process lock + atomic write	✅
delete_session()	in-process lock	n/a

Reads inside the lock guarantee that a metadata merge picks up any messages another worker appended just before. Since PR #1709, FileLock.__enter__ raises IOError on timeout (previously the lock failure was silent, which could lead to torn writes on a stuck lock). The default lock_timeout is 5.0 seconds — pass DefaultSessionStore(lock_timeout=...) to tune it.

Behaviour change in PR #1709: FileLock.__enter__ now raises IOError instead of returning silently on lock-acquisition timeout. If you have custom code that uses FileLock directly from praisonaiagents.session.store, wrap it in a try/except IOError or raise the timeout via DefaultSessionStore(lock_timeout=...).

Best Practices

Use agent methods

Call agent.chat() or agent.start() — these acquire locks internally.

response = agent.chat("Hello")

Avoid direct list mutation

Do not append to chat_history directly; use agent methods or the locked setter.

# Bad — bypasses locks
agent.chat_history.append({"role": "user", "content": "Hello"})

# Good
agent.chat("Hello")

Clear history safely

Use agent.clear_history() rather than assigning an empty list from multiple threads.

Serialise mixed sync/async callers

Sync and async locks are independent since PR #1567 — add an external lock when both contexts mutate history.

Async Considerations

agent.chat_history is async-aware out of the box — no external asyncio.Lock is required when all calls are inside the same event loop.

import asyncio
from praisonaiagents import Agent

agent = Agent(name="AsyncAgent")

async def async_chat(prompt):
    # No external lock needed - AsyncSafeState handles this
    return agent.chat(prompt)

async def main():
    tasks = [async_chat(f"Question {i}") for i in range(5)]
    results = await asyncio.gather(*tasks)

Since PR #1567, DualLock.sync() and DualLock.async_lock() use independent locks. A sync caller holding the lock will not block an async caller from acquiring it, and vice versa. Within a single context (all-sync or all-async) the lock works as expected; across contexts it does not coordinate. If you mutate agent.chat_history from both sync and async code paths, serialise the boundary yourself.

An external lock is still useful for serialising chat-history mutations from a thread pool that mixes sync and async callers:

import asyncio
import threading
from praisonaiagents import Agent

agent = Agent(name="MixedAgent")
external_lock = threading.Lock()

def sync_worker(prompt):
    with external_lock:
        return agent.chat(prompt)

async def async_worker(prompt):
    # Convert to sync context for coordination
    loop = asyncio.get_event_loop()
    with external_lock:
        return await loop.run_in_executor(None, agent.chat, prompt)

Verifying Thread Safety

Test thread safety with concurrent access:

import threading
from praisonaiagents.lite import LiteAgent

def test_thread_safety():
    agent = LiteAgent(
        name="Test",
        llm_fn=lambda m: "Response"
    )
    
    errors = []
    
    def worker():
        try:
            for _ in range(100):
                agent.chat("Test")
        except Exception as e:
            errors.append(e)
    
    threads = [threading.Thread(target=worker) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    
    assert len(errors) == 0, f"Thread safety errors: {errors}"
    print("Thread safety test passed!")

test_thread_safety()

Multi-team HTTP launch

PraisonAI provides comprehensive thread-safety for HTTP server deployment:

• Multiple Agent / Agents instances may call .launch(port=N) concurrently from different threads — registration is atomic.
• If two launch calls use the same path on the same port, the second gets an auto-suffixed path (/path_abc123) and a warning is logged.
• Server readiness is signalled deterministically (no fixed sleep); .launch() returns only after the port is accepting connections. The wait defaults to 5 seconds and is configurable via the PRAISONAI_SERVER_READY_TIMEOUT environment variable. If the server doesn’t become ready in time, .launch() still returns and a warning is logged — check server logs for startup errors.
• aworkflow() state lock is created inside the running async context, so workflows remain stable when invoked under pytest-asyncio or when nested inside another loop.

import threading
from praisonaiagents import AgentTeam

def launch_team(team_name, port, path):
    team = AgentTeam(name=team_name)
    team.launch(port=port, path=path)

# Safe concurrent launches
thread1 = threading.Thread(target=launch_team, args=("TeamA", 8000, "/team_a"))
thread2 = threading.Thread(target=launch_team, args=("TeamB", 8000, "/team_b"))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

# Both teams available at:
# - http://localhost:8000/team_a
# - http://localhost:8000/team_b

Wrapper-layer thread safety (praisonai package)

The praisonai wrapper layer (distinct from the praisonaiagents content above) provides thread-safe OpenAI client management and CLI command discovery.

Per-instance OpenAI client lifecycle

Each BaseAutoGenerator owns its own core OpenAIClient (from praisonaiagents.llm.openai_client), which manages sync and async access internally — no cross-instance sharing, no LRU eviction surprises.

import os
from praisonai import PraisonAI

# Each PraisonAI instance constructs its own auto-generator, which owns
# a single OpenAI client. Two concurrent instances cannot evict each
# other's client — even with identical API keys.
team_a = PraisonAI(auto="Draft a marketing plan", api_key=os.getenv("OPENAI_API_KEY"))
team_b = PraisonAI(auto="Draft a technical doc", api_key=os.getenv("OPENAI_API_KEY"))

team_a.run()  # safe to run in parallel threads
team_b.run()

The client is created lazily on first structured-completion call. __del__ was removed in PR #1736 and the canonical path is now explicit close() or aclose() on the generator, or — better — the context managers. This matters in long-lived server processes that spawn many generators.

For power users building generators directly:

import os
from praisonai.auto import BaseAutoGenerator

# Sync context manager
with BaseAutoGenerator(config_list=[{
    "model": "gpt-4o-mini",
    "api_key": os.getenv("OPENAI_API_KEY"),
    "base_url": None,
}]) as gen:
    # _get_core_client() is lock-guarded; calling it from multiple threads
    # on the same instance returns the same core OpenAIClient.
    result = gen._structured_completion(MyModel, messages=[...])

# Async context manager
async with BaseAutoGenerator(config_list=[{
    "model": "gpt-4o-mini",
    "api_key": os.getenv("OPENAI_API_KEY"),
    "base_url": None,
}]) as gen:
    result = await gen._astructured_completion(MyModel, messages=[...])

Behaviour change in PR #1681: the module-level functions praisonai.auto._get_openai_client(api_key, base_url) and the _openai_clients / _openai_clients_lock globals have been removed. If you imported them, switch to constructing an OpenAI client yourself or call BaseAutoGenerator(...).\_get_openai_client(). Each generator now owns exactly one client; the previous bug — an in-use client being evicted from a process-wide LRU and closed while other threads still held a reference — is no longer possible.

Behaviour change in PR #1736: __del__ was removed and async support was added. New methods include aclose, __aenter__/__aexit__, and _astructured_completion. Use context managers or explicit cleanup instead of relying on destructors.

Behaviour change in PR for #2963: BaseAutoGenerator now owns a single _core_client: OpenAIClient (the core-owned client) instead of separate _openai_client / _async_openai_client attributes. The methods _get_openai_client() and _get_async_openai_client() were consolidated into _get_core_client(). The public surface (close, aclose, __enter__/__exit__, __aenter__/__aexit__, _structured_completion, _astructured_completion) is unchanged. If you called the previous private methods directly, switch to _get_core_client().

Thread-safe Typer command discovery

Embedding python -m praisonai from multiple threads is now safe. The CLI command discovery uses a double-check lock pattern and doesn’t poison the cache on failure:

import threading
import subprocess

def run_cli_command(command):
    # Safe to call from multiple threads
    result = subprocess.run(
        ["python", "-m", "praisonai"] + command,
        capture_output=True, text=True
    )
    return result.stdout

# Multiple threads can safely use the CLI
threads = [
    threading.Thread(target=run_cli_command, args=(["--version"],))
    for _ in range(5)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

Failure-safe cache

A transient discovery error does not lock the CLI into a broken state — the next call retries instead of permanently breaking dispatch. This ensures reliable operation in multi-threaded server environments where temporary import failures might occur.

New Thread-Safe Components in PR #1548

AsyncAgentScheduler is now loop-aware. The start() method binds its async primitives (asyncio.Event, asyncio.Lock) to the running loop, and stop() raises RuntimeError if called from a different loop than start().

Lazy loaders in praisonai/auto.py are now thread-safe. A single _load_optional(key, loader) helper with a module-level lock replaces the previous unguarded module-level globals.

inbuilt_tools lazy import (PR #1681) now routes through praisonai.auto._load_optional("inbuilt_autogen_tools", ...) instead of a hand-rolled re-entry guard. Negative results are cached, so a missing crewai or autogen install no longer pays the find_spec cost on every attribute access.

Framework availability constants (PR #1780) — Module-level constants on praisonai.agents_generator (AGENTOPS_AVAILABLE, etc.) are resolved lazily via __getattr__. In praisonai.observability.hooks, the eager AGENTOPS_AVAILABLE constant was removed (PR #2062) — use is_agentops_available() instead.

Integration registry (praisonai/integrations/registry.py) now has a per-instance threading.Lock guarding register/unregister/create/list_registered operations.

New Thread-Safe Components in PR #1673

Jobs server singleton init (PR #1771) — get_store() and get_executor() in praisonai/jobs/server.py now use double-checked locking with threading.Lock. This eliminates a TOCTOU race where concurrent cold-start requests could create orphaned JobStore / JobExecutor instances on a fresh process.

InMemoryJobStore — locked reads and async get_stats()

All read methods (get, get_by_idempotency_key, list_jobs, count, get_stats) now hold an asyncio.Lock while reading internal dicts, so concurrent saves cannot tear a read.

Breaking change: get_stats() is now a coroutine. Update your code:

# Before PR #1673
stats = store.get_stats()

# After PR #1673
stats = await store.get_stats()

AgentScheduler — interruptible retry backoff (sync scheduler)

stop() now becomes responsive within milliseconds even during retry backoff. The sync scheduler also adopts the shared backoff_delay() curve so sync and async retries are identical.

from praisonai.scheduler import AgentScheduler

scheduler = AgentScheduler(agent, "Hourly news check")
scheduler.start("hourly", max_retries=3)
# ... later, from another thread or signal handler:
scheduler.stop()  # returns within ms, even if a backoff sleep was in progress

ToolRegistry — thread-safe registry operations

ToolRegistry now holds a threading.Lock around all reads and mutations, matching PluginRegistry / integration registry. Eliminates RuntimeError: dictionary changed size during iteration when registering tools concurrently with iteration. Reference: PR #1673.

Breaking change (PR #2248): The four AutoGen-specific methods that were added in PR #1780 (register_autogen_adapter, get_autogen_adapter, list_autogen_adapters, register_builtin_autogen_adapters) have been removed from ToolRegistry. Migrate to AutoGenAdapter from praisonai.framework_adapters or use praisonai.persistence.factory for persistence. See ToolRegistry AutoGen Migration for replacement patterns.

MCP registry lazy-loader locks (PR #2738)

All three MCP registries (MCPToolRegistry, MCPResourceRegistry, MCPPromptRegistry) now guard the lazy-loader queue with a threading.Lock using an atomic snapshot-then-run pattern. Concurrent list_all / get calls from stdio and HTTP transports are safe — each loader runs exactly once, regardless of caller count.

Each _ensure_loaded acquires the lock, snapshots _lazy_loaders, clears the pending set, releases the lock, then runs each loader outside the critical section:

Under 10-thread concurrency, loaders can no longer double-run or be skipped mid-iteration. Reference: PR #2738.

Multi-agent context safety (PR #1723)

praisonai.cli.features.interactive_tools._get_shared_runtime() and praisonai.tool_resolver._get_default_resolver() were previously process-wide singletons. They are now per-context via contextvars.ContextVar:

• Concurrent agents in the same process no longer share an InteractiveRuntime (no LSP/ACP config bleed).
• Resolvers anchor to each agent’s CWD, not whichever agent ran first.
• For long-lived daemons that switch projects, call praisonai.tool_resolver.reset_default_resolver() to force re-anchoring.

This eliminated the singleton-config-bleed and first-caller-CWD-wins bugs while keeping cleanup_runtime() and convenience functions like resolve_tool() source-compatible.

---

Related

Agent Cloning

Clone agents for channel isolation

Rate Limiter

Thread-safe rate limiting across agents