Recipe: Sync Wrappers and Sync Structured Calls¶
Goal¶
Use synchronous APIs where async orchestration is not desired.
Example (Agent.run_sync)¶
from doc_processing_toolkit import Agent
agent = Agent(
model="claude-sonnet-4-5",
instructions="Be concise.",
)
result = agent.run_sync("What is 2 + 2?")
print(result.output)
Calling from inside a running event loop¶
Agent.run_sync() and AgentSession.run_sync() work whether or not a loop is already running on
the calling thread:
- No loop running — the coroutine runs under
asyncio.run()directly. - A loop already running (Jupyter,
pytest-asyncio, a FastMCP tool handler executing on the server's own loop) — the coroutine is handed to a single worker thread with its own event loop, and the calling thread blocks until it returns.
Before 0.3.0 the second case raised RuntimeError: This event loop is already running unless the
caller installed nest_asyncio. The toolkit no longer uses nest_asyncio at all.
Local MCP servers opened by the run are connected and released inside that worker loop — one server
process per run_sync() call. To reuse a single server process across runs, keep a long-lived loop
and await agent.run(...) instead. See Tools.
Provider-level sync wrappers¶
Provider modules include sync wrapper variants for function-calling flows and native sync structured-output calls.
What to validate¶
- wrapper returns dict/result with the same contract as async path
- kwargs pass-through behaves as expected
- event-loop related errors are avoided in synchronous contexts, including when the caller already has a running loop
Notes¶
- For web backends and async workers, prefer async APIs and avoid sync wrappers in hot paths.
Test sources¶
tests/integration/test_sync_wrappers.py(all classes and sync wrapper scenarios)tests/e2e/test_agent_e2e.py(run_syncusage patterns)