> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    subgraph "Thread Safety"
        T1[🧵 Thread 1] --> Lock[🔒 AsyncSafeState]
        T2[🧵 Thread 2] --> Lock
        Lock --> History[💬 chat_history]
    end

    classDef thread fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef lock fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef shared fill:#10B981,stroke:#7C90A0,color:#fff

    class T1,T2 thread
    class Lock lock
    class History shared
```

## How It Works

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
    participant User
    participant Agent
    participant Feature as Thread-Safe Agent State

    User->>Agent: Request
    Agent->>Feature: Process request
    Feature-->>Agent: Result    Agent-->>User: Response
```

## Quick Start

<Steps>
  <Step title="Multi-threaded Chat">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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()
    ```
  </Step>

  <Step title="Async Concurrent Tasks">
    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    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())
    ```
  </Step>
</Steps>

<Warning>
  **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.

  ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  # 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
  ```
</Warning>

## 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:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

<Note>
  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
</Note>

#### What changed in PR #1514

<Note>
  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.
</Note>

#### Safe operations

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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](/docs/features/rate-limiter#thread-safety--multi-agent-use) for patterns.

## LiteAgent Thread Safety

The lite package also provides thread-safe operations:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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](https://github.com/MervinPraison/PraisonAI/pull/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.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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()
```

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

### 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.

<Warning>
  **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=...)`.
</Warning>

## Best Practices

<AccordionGroup>
  <Accordion title="Use agent methods">
    Call `agent.chat()` or `agent.start()` — these acquire locks internally.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    response = agent.chat("Hello")
    ```
  </Accordion>

  <Accordion title="Avoid direct list mutation">
    Do not append to `chat_history` directly; use agent methods or the locked setter.

    ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
    # Bad — bypasses locks
    agent.chat_history.append({"role": "user", "content": "Hello"})

    # Good
    agent.chat("Hello")
    ```
  </Accordion>

  <Accordion title="Clear history safely">
    Use `agent.clear_history()` rather than assigning an empty list from multiple threads.
  </Accordion>

  <Accordion title="Serialise mixed sync/async callers">
    Sync and async locks are independent since PR #1567 — add an external lock when both contexts mutate history.
  </Accordion>
</AccordionGroup>

## 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.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)
```

<Warning>
  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.
</Warning>

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    subgraph "Sync Context"
        SyncCaller[👨‍💻 Sync Caller] --> SyncLock[🔒 threading.RLock]
        SyncLock --> ChatHistory[💬 chat_history]
    end
    
    subgraph "Async Context"
        AsyncCaller[🚀 Async Caller] --> AsyncLock[🔄 asyncio.Lock<br/>per event loop]
        AsyncLock --> ChatHistory
    end
    
    SyncLock -.->|no cross-context<br/>coordination| AsyncLock
    
    classDef sync fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef async fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef shared fill:#10B981,stroke:#7C90A0,color:#fff
    
    class SyncCaller,SyncLock sync
    class AsyncCaller,AsyncLock async
    class ChatHistory shared
```

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

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    subgraph "Per-Instance OpenAI Client"
        G1[🤖 BaseAutoGenerator A] --> L1[🔒 Instance Lock] --> C1[📡 Client A]
        G2[🤖 BaseAutoGenerator B] --> L2[🔒 Instance Lock] --> C2[📡 Client B]
        G1 -.close().-> X1[♻️ client.close]
        G2 -.close().-> X2[♻️ client.close]
    end

    classDef generator fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef lock fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef client fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef cleanup fill:#10B981,stroke:#7C90A0,color:#fff

    class G1,G2 generator
    class L1,L2 lock
    class C1,C2 client
    class X1,X2 cleanup
```

### Per-instance OpenAI client lifecycle

Each `BaseAutoGenerator` owns its own OpenAI client — no cross-instance sharing, no LRU eviction surprises.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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()
```

<Note>
  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.
</Note>

For power users building generators directly:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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_openai_client() is double-checked-locked; calling it from
    # multiple threads on the same instance returns the same client.
    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=[...])
```

<Warning>
  **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.
</Warning>

### 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:

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

<Warning>
  **Breaking change:** `get_stats()` is now a coroutine. Update your code:

  ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
  # Before PR #1673
  stats = store.get_stats()

  # After PR #1673
  stats = await store.get_stats()
  ```
</Warning>

**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.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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](https://github.com/MervinPraison/PraisonAI/pull/1673).

<Warning>
  **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](/docs/features/tool-registry-autogen-migration) for replacement patterns.
</Warning>

### 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:

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    S[stdio caller] --> L[🔒 lock → snapshot → clear]
    H1[HTTP caller] --> L
    H2[HTTP caller 2] --> L
    L --> R[▶️ run each loader once]

    classDef caller fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef lock fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef run fill:#10B981,stroke:#7C90A0,color:#fff
    class S,H1,H2 caller
    class L lock
    class R run
```

Under 10-thread concurrency, loaders can no longer double-run or be skipped mid-iteration. Reference: [PR #2738](https://github.com/MervinPraison/PraisonAI/pull/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

<CardGroup cols={2}>
  <Card title="Agent Cloning" icon="copy" href="/docs/features/agent-cloning">
    Clone agents for channel isolation
  </Card>

  <Card title="Rate Limiter" icon="gauge" href="/docs/features/rate-limiter">
    Thread-safe rate limiting across agents
  </Card>
</CardGroup>
