Iter Coroutine

vercel/iter-coroutine

Use when refactoring vercel-py sync and async modules to share transport-agnostic business logic with the iter_coroutine pattern, including base/runtime splits and shared HTTP request clients.

3k tokens

context cost

the whole folder, loaded on every use

files

instructions only

copies elsewhere

how many repositories repackaged it

stars on the repo

on the repository, not the skill itself

Install

one command, takes just this skill from the repository

npx skills add https://github.com/vercel/vercel-py --skill iter-coroutine

The instruction itself

16 sections, as written by the author

Iter-Coroutine + Base/Runtime Migration

Refactor sync+async modules to eliminate duplication using the iter-coroutine

pattern. This skill covers the full migration workflow: identifying candidates,

structuring internal modules, writing transport-agnostic business logic, and

wiring up sync/async public APIs.

When to use

A module has parallel sync and async implementations with duplicated logic.
You're adding a new feature that needs both sync and async public APIs.
You're refactoring an existing feature to use the shared HTTP transport layer.

Core principles

Public API is stable. Same exported names, signatures, return types, and

behavior. Migrations are internal refactors, not API changes.

Internal core is async-first. All shared business logic lives in async

methods on a base class.

Sync entrypoints are thin wrappers over the async core via

iter_coroutine(...), used only when the wrapped coroutine is non-suspending

in sync mode.

HTTP goes through vercel._internal.http transports. Never construct

httpx.Client or httpx.AsyncClient directly in feature modules.

Architecture overview

Public API (sync wrappers + async functions)
    │
    │ iter_coroutine() one-shot bridge (sync path)
    │ await (async path)
    ▼
Concrete Clients (SyncFooClient / AsyncFooClient)
    │
    │ inherit from
    ▼
BaseFooClient (all async methods — shared business logic)
    │
    │ uses
    ▼
RequestClient (transport-agnostic async API)
    │
    ▼
Transport (SyncTransport or AsyncTransport)
    │
    ▼
httpx.Client or httpx.AsyncClient

How `iter_coroutine` works

# src/vercel/_internal/iter_coroutine.py
def iter_coroutine(coro: Coroutine[None, None, _T]) -> _T:
    """Execute a non-suspending coroutine synchronously."""
    try:
        coro.send(None)
    except StopIteration as ex:
        return ex.value
    else:
        raise RuntimeError(f"coroutine {coro!r} did not stop after one iteration!")
    finally:
        coro.close()

It drives a coroutine forward exactly one step. If the coroutine completes

without suspending, its return value is extracted from the StopIteration

exception. If it tries to yield/await, it raises RuntimeError.

Safety rules

| Safe | Unsafe |

|---------------------------------------------------|--------------------------------------------------|

| Coroutines that only call sync code via async API | Coroutines that await real I/O |

| SyncTransport.send() (sync under async facade) | AsyncTransport.send() (real await) |

| Sync sleep, sync file I/O | asyncio.sleep, asyncio.create_task |

| inspect.isawaitable() guarded branches | Unguarded await on user callbacks |

Step-by-step migration

1. Create the transport-agnostic request client

The request client wraps a BaseTransport and exposes an async API. The

transport implementation determines whether I/O is actually sync or async.

# src/vercel/_internal/foo/core.py
from vercel._internal.http import (
    AsyncTransport, BaseTransport, SyncTransport,
    create_base_client, create_base_async_client,
)

class FooRequestClient:
    def __init__(
        self,
        *,
        transport: BaseTransport,
        sleep_fn: SleepFn = asyncio.sleep,
        await_callback: bool = True,
    ) -> None:
        self._transport = transport
        self._sleep_fn = sleep_fn
        self._await_callback = await_callback

    async def request(self, method: str, path: str, **kwargs) -> Any:
        # Retry logic, error mapping, progress callbacks — all async-shaped
        # but non-suspending when backed by SyncTransport
        resp = await self._transport.send(method, path, **kwargs)
        ...

Factory functions configure transport + sleep + callback behavior:

def create_sync_request_client(timeout: float = 30.0) -> FooRequestClient:
    transport = SyncTransport(create_base_client(timeout=timeout))
    return FooRequestClient(
        transport=transport,
        sleep_fn=_sync_sleep,         # time.sleep, not asyncio.sleep
        await_callback=False,          # don't await user callbacks
    )

def create_async_request_client(timeout: float = 30.0) -> FooRequestClient:
    transport = AsyncTransport(create_base_async_client(timeout=timeout))
    return FooRequestClient(transport=transport)  # defaults are async-friendly

2. Write the shared base client

All business logic goes in async methods on a base class:

class BaseFooClient:
    def __init__(self, *, request_client: FooRequestClient) -> None:
        self._request_client = request_client

    async def do_thing(self, path: str, *, token: str) -> Result:
        # Validation, request building, response parsing — all here
        raw = await self._request_client.request("POST", path, token=token)
        return build_result(raw)

3. Add sync and async concrete clients

class SyncFooClient(BaseFooClient):
    def __init__(self) -> None:
        super().__init__(request_client=create_sync_request_client())

    def close(self) -> None:
        self._request_client.close()

    def __enter__(self) -> SyncFooClient:
        return self

    def __exit__(self, *args: object) -> None:
        self.close()


class AsyncFooClient(BaseFooClient):
    def __init__(self) -> None:
        super().__init__(request_client=create_async_request_client())

    async def aclose(self) -> None:
        await self._request_client.aclose()

    async def __aenter__(self) -> AsyncFooClient:
        return self

    async def __aexit__(self, *args: object) -> None:
        await self.aclose()

4. Wire up public API functions

# src/vercel/foo/ops.py
from vercel._internal.iter_coroutine import iter_coroutine

_T = TypeVar("_T")

def _run_sync(
    operation: Callable[[SyncFooClient], Coroutine[None, None, _T]],
) -> _T:
    with SyncFooClient() as client:
        return iter_coroutine(operation(client))


# Sync public API
def do_thing(path: str, *, token: str | None = None) -> Result:
    token = ensure_token(token)
    return _run_sync(lambda c: c.do_thing(path, token=token))


# Async public API
async def do_thing_async(path: str, *, token: str | None = None) -> Result:
    token = ensure_token(token)
    async with AsyncFooClient() as client:
        return await client.do_thing(path, token=token)

Base + runtime split

When sync and async paths differ materially (e.g., threading vs asyncio for

concurrent uploads), use a runtime object:

# Sync runtime — ThreadPoolExecutor
class _SyncUploadRuntime:
    def upload(self, *, parts, upload_fn, ...) -> list[PartResult]:
        with ThreadPoolExecutor(max_workers=MAX_CONCURRENCY) as executor:
            # Submit work, collect results
            ...

# Async runtime — anyio task group
class _AsyncUploadRuntime:
    async def upload(self, *, parts, upload_fn, ...) -> list[PartResult]:
        semaphore = anyio.Semaphore(MAX_CONCURRENCY)
        async with anyio.create_task_group() as tg:
            # Start tasks with semaphore limiting
            ...

The base client calls the runtime polymorphically:

class BaseFooClient:
    async def _do_upload(self, ...) -> list[PartResult]:
        parts = await _await_if_necessary(
            self._runtime.upload(
                parts=chunks,
                upload_fn=self._make_upload_fn(),
                ...
            )
        )
        return order_parts(parts)

Concrete clients inject the right runtime and upload function:

class SyncFooClient(BaseFooClient):
    def __init__(self) -> None:
        super().__init__(runtime=create_sync_runtime())

    def _make_upload_fn(self):
        # Wrap async method with iter_coroutine for sync execution
        return lambda **kw: iter_coroutine(self._client.upload_part(**kw))


class AsyncFooClient(BaseFooClient):
    def __init__(self) -> None:
        super().__init__(runtime=create_async_runtime())

    def _make_upload_fn(self):
        # Return the async method directly
        return self._client.upload_part

Shared orchestration

Keep these in the base layer, not duplicated per runtime:

Input validation
Chunk/part iteration helpers
Result normalization and ordering
Final response shaping

Callback handling

For callbacks that may be sync or async:

async def _emit_progress(
    callback: Callable[[Event], None] | Callable[[Event], Awaitable[None]] | None,
    event: Event,
    *,
    await_callback: bool,
) -> None:
    if callback is None:
        return
    result = callback(event)
    if await_callback and inspect.isawaitable(result):
        await cast(Awaitable[None], result)

Sync clients set await_callback=False — callbacks are always sync.
Async clients set await_callback=True — callbacks may be awaitable.

Testing expectations

Prefer integration-style tests with respx that verify real request flow and

sync/async parity.

Do not rely only on monkeypatch tests that assert internal call shape.
Test both sync and async paths for every operation.
Validate before commit:
./scripts/lint.sh
./scripts/test.sh (targeted or full)

Reference implementation

The blob module is the canonical example of this pattern:

Transport: src/vercel/_internal/http/transport.py
Request client: src/vercel/_internal/blob/core.py → BlobRequestClient
Base client: src/vercel/_internal/blob/core.py → BaseBlobOpsClient
Sync client: src/vercel/_internal/blob/core.py → SyncBlobOpsClient
Async client: src/vercel/_internal/blob/core.py → AsyncBlobOpsClient
Runtime split: src/vercel/_internal/blob/multipart.py
Public API: src/vercel/blob/ops.py
Types re-export: src/vercel/blob/types.py

How to use it

Copy the folder

Take vercel/iter-coroutine from the repository into ~/.claude/skills for personal use, or into .claude/skills inside a project.

Check the name does not clash

The agent identifies a skill by the name field in its header. Two skills with the same name cannot sit side by side — one of them will be ignored.