Prompt
When working with concurrency (threads/async), ensure code is safe under parallel invocations and does not block or re-enter running event loops.
Rules: 1) Eliminate shared mutable state across requests/threads
- Prefer immutable middleware/config objects.
- Never store request-specific data in process-global or shared containers unless it is explicitly per-thread/per-run.
2) Be event-loop safe
- Never call
asyncio.run()/loop.run_until_complete()when a loop may already be running. - Provide real async APIs for IO-bound work; for sync entrypoints, delegate blocking work to an executor.
3) Provide and test async behavior
- If production uses async, add unit tests for async paths.
- Add coverage for parallel execution (e.g., parallel tool calls).
4) Keep async detection consistent
- Prefer
inspect.iscoroutinefunctionoverasyncio.iscoroutinefunctionfor newer Python/mypy compatibility.
Example patterns:
# 1) Immutability / no post-init mutation
class ConditionalModelSettingsMiddleware(AgentMiddleware):
def __init__(self, conditions: list[tuple[Callable, dict[str, Any]]]):
super().__init__()
self._conditions = list(conditions) # don’t mutate after init
# 2) Offload blocking work from async contexts
async def abefore_agent(self, state, runtime):
return await run_in_executor(None, self.before_agent, state, runtime)
# 3) Don’t use asyncio.run() inside possibly-running event loops
# Prefer: make the validation async, or detect running loop and schedule appropriately.
async def _avalidate(...):
...
# then in async contexts: await _avalidate(...)
# 4) Prefer inspect for coroutine-function checks
import inspect
if ignore_condition_name is None or not getattr(handler, ignore_condition_name):
event = getattr(handler, event_name)
if inspect.iscoroutinefunction(event):
...
Adopting these practices prevents race conditions, avoids event-loop reentrancy problems, and ensures concurrency-heavy logic is correct under real async/parallel workloads.