Skills Asynchronous and Concurrent Processing

When a Skill needs to wait for multiple time-consuming operations simultaneously (such as network requests, file I/O), asynchronous programming can significantly reduce total waiting time.

This article introduces the application scenarios and coding patterns of Python asyncio in Skill scripts.

Synchronous vs Asynchronous: When to Use Asynchronous

Scenario	Recommended Approach	Reason
Single file processing, single API call	Synchronous	Asynchronous benefits are not obvious; code is simpler
Send multiple HTTP requests simultaneously	Asynchronous	Concurrent waiting, total time ≈ longest single request time
Read multiple files simultaneously	Asynchronous or thread pool	Tasks can be switched while waiting for I/O
CPU-intensive computation (e.g., image processing)	Multiprocessing	asyncio does not bypass the GIL

Asynchronous does not equal parallel. Python's asyncio is a single-threaded event loop, suitable for I/O-intensive tasks (waiting for network, disk). CPU-intensive tasks should useProcessPoolExecutor。

Asynchronously Send Multiple HTTP Requests

Suppose a Skill needs to call 3 API endpoints simultaneously. In synchronous mode, the total time is the sum of the three requests; in asynchronous mode, the total time is approximately equal to the slowest one.

Example

# File path: scripts/async_requests.py
# Asynchronously and concurrently call multiple APIs and aggregate the results

import asyncio
import aiohttp # pip install aiohttp --break-system-packages
import json
import sys
import time

API_KEY = "your_api_key"
BASE_URL = "https://api.example.com/v1"

async def fetch_one(session: aiohttp.ClientSession,
endpoint: str, payload: dict) -> dict:
"""Asynchronously send a single POST request"""
url = BASE_URL + endpoint
headers = {"Authorization": f"Bearer {API_KEY}"}
try:
async with session.post(url, json=payload,
headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as resp:
resp.raise_for_status()
return {"endpoint": endpoint, "status": "ok", "data": await resp.json()}
except Exception as e:
return {"endpoint": endpoint, "status": "error", "message": str(e)}

async def fetch_all(requests: list) -> list:
"""
Send all requests concurrently and wait for all to complete

Parameters:
requests: list, each item is an (endpoint, payload) tuple
"""
async with aiohttp.ClientSession() as session:
tasks = [
fetch_one(session, endpoint, payload)
for endpoint, payload in requests
]
# gather concurrently executes all tasks; return_exceptions=True prevents one failure from affecting others
return await asyncio.gather(*tasks, return_exceptions=True)

def main():
# Define the list of requests to be called concurrently
requests = [
("/summarize", {"text": "Example is a technology learning platform", "lang": "zh"}),
("/keywords", {"text": "EXAMPLE provides various programming tutorials", "top_n": 5}),
("/sentiment", {"text": "The tutorials on this platform are of high quality"}),
]

t0 = time.perf_counter()
results = asyncio.run(fetch_all(requests))
elapsed = time.perf_counter() - t0

print(f"Completed {len(results)} requests concurrently, total time {elapsed:.2f} seconds")
print(json.dumps(results, ensure_ascii=False, indent=2))

if __name__ == "__main__":
main()

并发完成 3 个请求，总耗时 1.23 秒
（同步串行约需 3.5 秒，并发节省约 65% 时间）

Asynchronously Read Multiple Files

For scenarios where multiple files need to be read simultaneously,aiofilesprovides an asynchronous version of file I/O.

Example

# File path: scripts/async_file_reader.py
import asyncio
import aiofiles # pip install aiofiles --break-system-packages
import os

async def read_file(file_path: str) -> dict:
"""Asynchronously read a single file"""
try:
async with aiofiles.open(file_path, encoding="utf-8") as f:
content = await f.read()
return {
"file": os.path.basename(file_path),
"chars": len(content),
"lines": content.count("\n"),
"status": "ok"
}
except Exception as e:
return {"file": file_path, "status": "error", "message": str(e)}

async def read_all_files(file_paths: list) -> list:
"""Concurrently read all files"""
tasks = [read_file(fp) for fp in file_paths]
return await asyncio.gather(*tasks)

if __name__ == "__main__":
import glob, json
# Read all .txt files in the upload directory
files = glob.glob("/mnt/user-data/uploads/*.txt")
if not files:
print("No .txt files found")
else:
results = asyncio.run(read_all_files(files))
print(json.dumps(results, ensure_ascii=False, indent=2))

Timeout Control and Task Cancellation

In asynchronous tasks, timeout control is particularly important.asyncio.wait_forYou can set a maximum waiting time for a single task.

Example

# File path: scripts/async_timeout.py
import asyncio

async def slow_task(name: str, delay: float) -> str:
"""Simulate a time-consuming task"""
await asyncio.sleep(delay)
return f"{name} completed"

async def run_with_timeout():
tasks = {
"Fast task": slow_task("Fast task", 0.5),
"Slow task": slow_task("Slow task", 5.0),
}
results = {}
for name, coro in tasks.items():
try:
# Wait at most 2 seconds for each task
result = await asyncio.wait_for(coro, timeout=2.0)
results[name] = {"status": "ok", "result": result}
except asyncio.TimeoutError:
results[name] = {"status": "timeout", "message": "Not completed within 2 seconds"}
return results

if __name__ == "__main__":
import json
output = asyncio.run(run_with_timeout())
print(json.dumps(output, ensure_ascii=False, indent=2))

{
  "快速任务": {"status": "ok",      "result": "快速任务 完成"},
  "慢速任务": {"status": "timeout", "message": "超过 2 秒未完成"}
}

Explain How to Invoke Asynchronous Scripts in SKILL.md

## 并发 API 调用

当用户提供多个文本需要同时分析时，运行异步脚本以提升速度：

```bash
python scripts/async_requests.py
```

该脚本会并发发出所有请求，总耗时约等于最慢的单次请求，
而非所有请求的时间之和。

Other Extensions

Skills Advanced

Skills Practical Projects

Skills Asynchronous and Concurrent Processing

Synchronous vs Asynchronous: When to Use Asynchronous

Asynchronously Send Multiple HTTP Requests

Example

Asynchronously Read Multiple Files

Example

Timeout Control and Task Cancellation

Example

Explain How to Invoke Asynchronous Scripts in SKILL.md