Async vs Sync

openocd-python is async-first: every subsystem is implemented as an async class using asyncio. For callers who do not need or want async, a complete set of synchronous wrappers is provided. The two APIs have identical functionality — the sync wrappers simply call run_until_complete() on the underlying async methods.

The two API surfaces

Every subsystem exists as a pair:

Async class	Sync wrapper
Session	SyncSession
Target	SyncTarget
Memory	SyncMemory
Registers	SyncRegisters
Flash	SyncFlash
JTAGController	SyncJTAGController
BreakpointManager	SyncBreakpointManager
SVDManager	SyncSVDManager

The async classes are the primary implementation. The Sync* wrappers delegate every method call through an event loop using loop.run_until_complete().

Async usage

Use the async API when:

• You are inside an async def function already
• You are building on top of an async framework (FastAPI, aiohttp, etc.)
• You need to run multiple OpenOCD operations concurrently
• You are integrating with other async I/O (serial ports, network services, etc.)

import asyncio
from openocd import Session

async def main():
    async with Session.connect() as ocd:
        # All subsystem methods use await
        state = await ocd.target.state()
        print(f"State: {state.state}")

        if state.state == "halted":
            pc = await ocd.registers.pc()
            dump = await ocd.memory.hexdump(pc, 32)
            print(dump)

        await ocd.target.resume()

asyncio.run(main())

Async with FastAPI

from fastapi import FastAPI
from openocd import Session

app = FastAPI()

@app.get("/target/state")
async def get_target_state():
    async with Session.connect() as ocd:
        state = await ocd.target.state()
        return {
            "name": state.name,
            "state": state.state,
            "pc": state.current_pc,
        }

Sync usage

Use the sync API when:

• You are writing a simple script
• You are working in a REPL or Jupyter notebook
• Your codebase is synchronous
• You do not need concurrent I/O

from openocd import Session

with Session.connect_sync() as ocd:
    # No await needed -- methods block until complete
    state = ocd.target.state()
    print(f"State: {state.state}")

    if state.state == "halted":
        pc = ocd.registers.pc()
        dump = ocd.memory.hexdump(pc, 32)
        print(dump)

    ocd.target.resume()

The sync entry points are Session.connect_sync() and Session.start_sync(). They return a SyncSession instead of a Session.

How the sync wrapper works

When you call Session.connect_sync(), three things happen:

1. Event loop creation: _get_or_create_loop() gets or creates an asyncio event loop. If there is no running loop, it uses the existing one (or creates a new one). If there is already a running loop, it raises a RuntimeError.

2. Async method execution: The async Session.connect() is run via loop.run_until_complete().

3. SyncSession wrapping: The resulting Session is wrapped in a SyncSession that stores both the session and the loop.

Every method on SyncSession (and the Sync* subsystem wrappers) follows the same pattern:

# Inside SyncTarget
def halt(self) -> TargetState:
    return self._loop.run_until_complete(self._target.halt())

This is straightforward delegation — no additional logic, no caching, no threading.

The async context guard

Caution

Never call the sync API from inside an already-running async context. The _get_or_create_loop() function detects this and raises a RuntimeError to prevent deadlocks.

The guard works by checking for an active event loop:

def _get_or_create_loop() -> asyncio.AbstractEventLoop:
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        pass  # No running loop -- safe to proceed
    else:
        raise RuntimeError(
            "Cannot use sync API from an async context. "
            "Use the async Session.start()/connect() instead."
        )
    # ... create or get event loop

This means the following will fail:

import asyncio
from openocd import Session

async def bad_idea():
    # This raises RuntimeError!
    with Session.connect_sync() as ocd:
        ocd.target.state()

asyncio.run(bad_idea())

The fix is to use the async API inside async contexts:

import asyncio
from openocd import Session

async def correct():
    async with Session.connect() as ocd:
        await ocd.target.state()

asyncio.run(correct())

When to use which

Scenario	API	Why
Simple automation script	Sync	Less boilerplate, no asyncio.run() needed
Jupyter notebook	Sync	Notebooks have their own event loop complications
pytest test (sync)	Sync	Straightforward test functions
pytest test (async)	Async	With pytest-asyncio and asyncio_mode = "auto"
FastAPI / aiohttp endpoint	Async	Already in an async context
MCP server tool	Async	FastMCP tools are async
Concurrent multi-target	Async	asyncio.gather() across multiple sessions
CI/CD flash script	Sync	Simple, linear flow

Mixing async and sync

You cannot mix the two styles within a single session. A Session is always used with await, and a SyncSession is always used without. However, you can have separate sessions of different types in the same program, as long as the sync calls happen outside any running event loop:

import asyncio
from openocd import Session

# Sync session for quick setup
with Session.connect_sync() as ocd:
    ocd.target.halt()
    ocd.target.reset(mode="halt")

# Async session for complex operations
async def do_work():
    async with Session.connect() as ocd:
        state = await ocd.target.state()
        # ... more async work

asyncio.run(do_work())

Type differences

The async and sync APIs return identical data types. TargetState, Register, FlashBank, TAPInfo, and all other dataclasses are shared between both APIs. The only difference is at the session and subsystem level:

# Async
ocd: Session
ocd.target       # -> Target
ocd.memory        # -> Memory
ocd.registers     # -> Registers

# Sync
ocd: SyncSession
ocd.target       # -> SyncTarget
ocd.memory        # -> SyncMemory
ocd.registers     # -> SyncRegisters

The return types of methods are identical:

# Both return TargetState
state = await ocd.target.state()   # async
state = ocd.target.state()         # sync

# Both return list[int]
words = await ocd.memory.read_u32(0x08000000, count=4)  # async
words = ocd.memory.read_u32(0x08000000, count=4)        # sync

Next steps

• Session Lifecycle — how sessions are created and torn down
• Error Handling — exception handling works the same in both APIs
• Target Control — examples showing both async and sync patterns