Is your feature request related to a specific problem?
Currently, all local and GCS skill loading and listing functions (load_skill_from_dir, load_skill_from_gcs_dir, list_skills_in_dir, list_skills_in_gcs_dir) are strictly synchronous and perform blocking disk or network I/O.
While the SkillRegistry abstract interface uses async methods, implementing a registry that loads local/GCS skills or using these loading utilities inside an asynchronous web server (like FastAPI) or async runners causes the main event loop to block, degrading performance and responsiveness.
Describe the Solution You'd Like
Add asynchronous counterparts for all 4 core skill loading/listing functions in the google.adk.skills package, preserving the existing synchronous functions for backward compatibility:
load_skill_from_dir_async(skill_dir)
load_skill_from_gcs_dir_async(bucket_name, skill_id, ...)
list_skills_in_dir_async(skills_base_path)
list_skills_in_gcs_dir_async(bucket_name, ...)
These functions should utilize Python's built-in asyncio.to_thread to offload blocking I/O operations to a background thread pool, keeping the async event loop completely non-blocking.
Impact on your work
Currently, the SkillRegistry abstract interface uses async def get_skill(...). However, actually loading a local or GCS skill inside registry implementations or in an async web server (like FastAPI) requires calling blocking synchronous functions (load_skill_from_dir), which freezes the entire asyncio event loop.
Adding these non-blocking async counterparts allows us to write fully async local and GCS registries and integrate them safely into concurrent web servers and runners without any thread-blocking performance bottlenecks.
Willingness to contribute
Are you interested in implementing this feature yourself or submitting a PR?
Yes, I have already implemented the code and comprehensive unit tests, and I am ready to submit a PR immediately.
Describe Alternatives You've Considered
Converting existing functions to async def: This was rejected as it would be a major breaking change for all existing synchronous callers (such as get_bigquery_skill() or synchronous user scripts). Providing new _async counterparts preserves 100% backward compatibility.
Using third-party async I/O libraries (e.g., aiofiles): This was rejected to avoid introducing heavy new dependencies to the ADK core package. Using Python's built-in asyncio.to_thread to wrap the highly tested synchronous functions is clean, lightweight, and extremely robust.
Proposed API / Implementation
Usage Example:
import asyncio
import pathlib
from google.adk import Agent
from google.adk.skills import load_skill_from_dir_async
from google.adk.tools import skill_toolset
# 1. Define a dummy tool
def get_weather(location: str) -> str:
return f"The current weather in {location} is 22°C and Sunny."
async def main():
# 2. Load the directory-based skill asynchronously (non-blocking!)
SKILLS_DIR = pathlib.Path(__file__).parent / "skills" / "weather_skill"
weather_skill = await load_skill_from_dir_async(SKILLS_DIR)
# 3. Create the SkillToolset
my_skill_toolset = skill_toolset.SkillToolset(
skills=[weather_skill],
additional_tools=[get_weather],
)
# 4. Initialize Agent
root_agent = Agent(
model="gemini-2.5-flash",
name="weather_agent",
tools=[my_skill_toolset],
)
# 5. Execute the agent asynchronously (non-blocking!)
response = await root_agent.achat("What's the weather like in Tokyo?")
print(response.text)
if __name__ == "__main__":
# Start the event loop and run our async main function
asyncio.run(main())
Additional Context
Add any other context or screenshots about the feature request here.
I have verified this implementation by adding 4 new async tests to tests/unittests/skills/test__utils.py. All tests pass successfully:
Is your feature request related to a specific problem?
Currently, all local and GCS skill loading and listing functions (load_skill_from_dir, load_skill_from_gcs_dir, list_skills_in_dir, list_skills_in_gcs_dir) are strictly synchronous and perform blocking disk or network I/O.
While the SkillRegistry abstract interface uses async methods, implementing a registry that loads local/GCS skills or using these loading utilities inside an asynchronous web server (like FastAPI) or async runners causes the main event loop to block, degrading performance and responsiveness.
Describe the Solution You'd Like
Add asynchronous counterparts for all 4 core skill loading/listing functions in the google.adk.skills package, preserving the existing synchronous functions for backward compatibility:
load_skill_from_dir_async(skill_dir)
load_skill_from_gcs_dir_async(bucket_name, skill_id, ...)
list_skills_in_dir_async(skills_base_path)
list_skills_in_gcs_dir_async(bucket_name, ...)
These functions should utilize Python's built-in asyncio.to_thread to offload blocking I/O operations to a background thread pool, keeping the async event loop completely non-blocking.
Impact on your work
Currently, the SkillRegistry abstract interface uses async def get_skill(...). However, actually loading a local or GCS skill inside registry implementations or in an async web server (like FastAPI) requires calling blocking synchronous functions (load_skill_from_dir), which freezes the entire asyncio event loop.
Adding these non-blocking async counterparts allows us to write fully async local and GCS registries and integrate them safely into concurrent web servers and runners without any thread-blocking performance bottlenecks.
Willingness to contribute
Are you interested in implementing this feature yourself or submitting a PR?
Yes, I have already implemented the code and comprehensive unit tests, and I am ready to submit a PR immediately.
Describe Alternatives You've Considered
Converting existing functions to async def: This was rejected as it would be a major breaking change for all existing synchronous callers (such as get_bigquery_skill() or synchronous user scripts). Providing new _async counterparts preserves 100% backward compatibility.
Using third-party async I/O libraries (e.g., aiofiles): This was rejected to avoid introducing heavy new dependencies to the ADK core package. Using Python's built-in asyncio.to_thread to wrap the highly tested synchronous functions is clean, lightweight, and extremely robust.
Proposed API / Implementation
Usage Example:
Additional Context
Add any other context or screenshots about the feature request here.
I have verified this implementation by adding 4 new async tests to tests/unittests/skills/test__utils.py. All tests pass successfully: