Python asyncio - Async/Await Concurrency

Overview

Python's asyncio library enables writing concurrent code using async/await syntax. It's ideal for I/O-bound operations like HTTP requests, database queries, file operations, and WebSocket connections. asyncio provides non-blocking execution without the complexity of threading or multiprocessing.

Key Features:

• async/await syntax for readable concurrent code
• Event loop for managing concurrent operations
• Tasks for running multiple coroutines concurrently
• Primitives: locks, semaphores, events, queues
• HTTP client/server with aiohttp
• Database async support (asyncpg, aiomysql, motor)
• FastAPI async endpoints
• WebSocket support
• Background task management

Installation:

# asyncio is built-in (Python 3.7+)

# Async HTTP client
pip install aiohttp

# Async HTTP requests (alternative)
pip install httpx

# Async database drivers
pip install asyncpg aiomysql motor  # PostgreSQL, MySQL, MongoDB

# FastAPI with async support
pip install fastapi uvicorn[standard]

# Async testing
pip install pytest-asyncio

Basic Async/Await Patterns

1. Simple Async Function

import asyncio

async def hello():
    """Basic async function (coroutine)."""
    print("Hello")
    await asyncio.sleep(1)  # Async sleep (non-blocking)
    print("World")
    return "Done"

# Run async function
result = asyncio.run(hello())
print(result)  # "Done"

Key Points:

• async def defines a coroutine function
• await suspends execution until awaitable completes
• asyncio.run() is the entry point for async programs
• Coroutines must be awaited or scheduled

2. Multiple Concurrent Tasks

import asyncio
import time

async def task(name, duration):
    """Simulate async task."""
    print(f"{name}: Starting (duration: {duration}s)")
    await asyncio.sleep(duration)
    print(f"{name}: Complete")
    return f"{name} result"

async def run_concurrent():
    """Run multiple tasks concurrently."""
    start = time.time()

    # Sequential (slow) - 6 seconds total
    # result1 = await task("Task 1", 3)
    # result2 = await task("Task 2", 2)
    # result3 = await task("Task 3", 1)

    # Concurrent (fast) - 3 seconds total
    results = await asyncio.gather(
        task("Task 1", 3),
        task("Task 2", 2),
        task("Task 3", 1)
    )

    elapsed = time.time() - start
    print(f"Total time: {elapsed:.2f}s")
    print(f"Results: {results}")

asyncio.run(run_concurrent())
# Output: Total time: 3.00s (tasks ran concurrently)

3. Task Creation and Management

import asyncio

async def background_task(name):
    """Long-running background task."""
    for i in range(5):
        print(f"{name}: iteration {i}")
        await asyncio.sleep(1)
    return f"{name} complete"

async def main():
    # Create task (starts immediately)
    task1 = asyncio.create_task(background_task("Task-1"))
    task2 = asyncio.create_task(background_task("Task-2"))

    # Do other work while tasks run
    print("Main: doing other work")
    await asyncio.sleep(2)

    # Wait for tasks to complete
    result1 = await task1
    result2 = await task2

    print(f"Results: {result1}, {result2}")

asyncio.run(main())

4. Error Handling in Async Code

import asyncio

async def risky_operation(fail=False):
    """Operation that might fail."""
    await asyncio.sleep(1)
    if fail:
        raise ValueError("Operation failed")
    return "Success"

async def handle_errors():
    # Individual try/except
    try:
        result = await risky_operation(fail=True)
    except ValueError as e:
        print(f"Caught error: {e}")
        result = "Fallback value"

    # Gather with error handling
    results = await asyncio.gather(
        risky_operation(fail=False),
        risky_operation(fail=True),
        risky_operation(fail=False),
        return_exceptions=True  # Return exceptions instead of raising
    )

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Task {i} failed: {result}")
        else:
            print(f"Task {i} succeeded: {result}")

asyncio.run(handle_errors())

Event Loop Fundamentals

1. Event Loop Lifecycle

import asyncio

# Modern approach (Python 3.7+)
async def main():
    print("Main coroutine")
    await asyncio.sleep(1)

asyncio.run(main()
how to use asyncio
CursorClaude CodeClineCodexWindsurfGooseGitHub CopilotVS CodeGemini CLIKiloOpencodeKimi CLIRoo ClineAiderContinueZedBolt.newLovablev0Replit AgentSweepSmol DeveloperDevinCavemanAmpAntigravity
How to use asyncio on Cursor
AI-first code editor with Composer
1
Prerequisites
Before installing skills in Cursor, ensure your development environment meets these requirements:
›Cursor installed and configured on your development machine
›Node.js version 16.0+ with npm package manager (verify with node --version)
›Active project directory or workspace where you want to add asyncio
2
Execute installation command
Execute the skills CLI command in your project's root directory to begin installation:
$npx skills add https://github.com/bobmatnyc/claude-mpm-skills --skill asynciocopy
The skills CLI fetches asyncio from GitHub repository bobmatnyc/claude-mpm-skills and configures it for Cursor.
3
Select Cursor when prompted
The CLI will show a list of available agents. Use arrow keys to navigate and space to select Cursor:
◆ Which agents do you want to install to?
│
│ ── Universal (.agents/skills) ── always included ────
│   • Amp
│   • Antigravity
│   • Cline
│   • Codex
│ ●Cursor(selected)
│   • Cursor
│   • Windsurf
4
Verify installation
Confirm successful installation by checking the skill directory location:
.cursor/skills/asyncio
Reload or restart Cursor to activate asyncio. Access the skill through slash commands (e.g., /asyncio) or your agent's skill management interface.
⚠
Security & Verification Notice
We perform automated surface-level scans (Gen AI Scanner, Socket, Snyk) during installation. These checks detect common vulnerabilities but do not guarantee complete security. Always review skill source code and verify the publisher's reputation before production use.
Skills execute code in your development environment. Always verify the publisher's identity, review recent commits, and test in isolated environments before production deployment.
Additional Resources
›View source code on GitHub›Skills CLI documentation›Learn more about Cursor›What are agent skills?