Generative AI on Google Cloud

27 / 80·notebook

Chapter 27

Intro to Skill Registry

NotebookPython 329 cells

In [ ]python · cell 1

python

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Intro to Skill Registry

Authors
Shirong Bai
Deep Dubey
Eric Dong

Overview

The Skill Registry is a secure, private repository for storing, indexing, and dynamically discovering agent skills on the Gemini Enterprise Agent Platform.

A skill extends an agent's capabilities by providing precise system instructions, domain-specific documentation, and local or remote tools. By indexing skills in the Registry, agents can perform semantic search at runtime to dynamically discover and load the exact capabilities needed to solve a user's request, optimizing context window usage and enforcing robust access control.

Objectives

In this notebook, you will learn how to:

Define a Custom Skill: Create a local skill package structure with instructions and metadata.
Upload and Register a Skill: Programmatically register custom skills in the Skill Registry.
Ingest Skills from GitHub: Batch-import ready-to-use agent skills directly from a GitHub repository.
Perform Semantic Discovery: Retrieve and search relevant skills dynamically using semantic query matching.

Getting Started

Install libraries

In [ ]python · cell 7

python

%pip install --upgrade --quiet "google-cloud-aiplatform==1.152.0"

⚠️ Note: Ignore pip dependency errors.

Import libraries

In [ ]python · cell 10

python

import os
import re
import sys
import shutil
import tempfile
import urllib.request
import yaml
import zipfile
import vertexai

from datetime import datetime

Authenticate your notebook environment

If you are running this notebook in Google Colab, execute the cell below to authenticate.

In [ ]python · cell 12

python

if "google.colab" in sys.modules:
    from google.colab import auth

    auth.authenticate_user()

Autenticate your Google Cloud project

You can use a Google Cloud Project or an API Key for authentication. This tutorial uses a Google Cloud Project.

Enable the Agent Platform API

In [ ]python · cell 14

python

# fmt: off
PROJECT_ID = "[your-project-id]"  # @param {type: "string", placeholder: "[your-project-id]", isTemplate: true}
# fmt: on
if not PROJECT_ID or PROJECT_ID == "[your-project-id]":
    PROJECT_ID = str(os.getenv("GOOGLE_CLOUD_PROJECT"))

LOCATION = "us-central1"

Initialize the Agent Platform client

Initialize the SDK and instantiate the central platform client. This client is the entry point for managing custom resources (such as agents and skills) on the Gemini Enterprise Agent Platform.

In [ ]python · cell 16

python

vertexai.init(project=PROJECT_ID, location=LOCATION)

client = vertexai.Client(project=PROJECT_ID, location=LOCATION)

Define and Register a Custom Skill

To register a skill in the Skill Registry, you need a skill package. At its simplest, a skill package is a local directory containing a mandatory SKILL.md file.

The SKILL.md file follows a specific layout:

YAML Frontmatter: A metadata block at the top specifying:
- name: A unique identifier matching the skill package name.
- description: A capability statement (starting in third-person) explaining what the skill does.
Instructions: The core markdown instructions that guide the agent's behavior when executing the skill.

Create a sample skill package

The following cell programmatically creates a temporary directory (/tmp/sample_math_skill) containing a sample SKILL.md file for a simple math helper.

In [ ]python · cell 18

python

# Create a temporary directory to simulate a local skill package
local_skill_dir = "/tmp/sample_math_skill"
os.makedirs(local_skill_dir, exist_ok=True)

# Create the SKILL.md file which defines the skill
skill_md_content = """---
name: sample-math-skill
description: A sample skill that helps with simple math addition. Use this skill when you need to add two numbers.
---
# Test Math Skill
This skill provides instructions on how to add two numbers.

## Instructions
To add two numbers, sum them up. For example, 2 + 2 = 4.
"""

with open(os.path.join(local_skill_dir, "SKILL.md"), "w") as f:
    f.write(skill_md_content)

print(f"Created sample skill directory at: {local_skill_dir}")

Upload and register the skill package

Now, upload and register your custom skill in the Skill Registry using client.skills.create.

The SDK will automatically package, compress, and upload the local skill folder specified in "local_path".
By default, the operation blocks synchronously (wait_for_completion=True) until the backend has successfully provisioned and indexed the skill.

In [ ]python · cell 20

python

ts = datetime.now().strftime("%Y%m%d-%H%m%S")
user_skill_id = f"math-skill-{ts}"

try:
    # Call create with local_path in config
    # By default, wait_for_completion is True, so this will block until the skill is created
    skill = client.skills.create(
        display_name="Sample math skill",
        description="This skill provides functions to perform math calculations",
        config={
            "skill_id": user_skill_id,
            "local_path": "/tmp/sample_math_skill"
        }
        )
    print(f"SUCCESS: Skill created successfully!")
    print(f"Skill Name: {skill.name}")
    print(f"Skill State: {skill.state}")
    print(f"Skill Display Name: {skill.display_name}")
    print(f"Skill Description: {skill.description}")
except Exception as e:
    print(f"FAILED: Skill creation failed: {e}")
    skill = None

Batch Ingestion of Skills from a GitHub Repository

In real-world production environments, organization-wide agent skills are often stored and managed in a central Git repository.

This section demonstrates how to build an automated ingestion pipeline to fetch, parse, and batch-register skills from a remote repository.

We will use the Google-managed google/skills repository as our input. This repository contains ready-to-use agent skills, such as:

bigquery-basics: Fundamental tools and instructions for querying Google Cloud BigQuery datasets.
cloud-run-basics: Key operations and deployment guidelines for managing Google Cloud Run services.

Once you run the pipeline with this repository as input, all contained skill packages will be programmatically ingested and registered inside your private Skill Registry, making them instantly discoverable.

In [ ]python · cell 22

python

GITHUB_REPO_URL = "https://github.com/google/skills" # @param {type:"string"}

Define pipeline helper functions

Define the necessary Python helper functions to automate the ingestion workflow. This includes downloading the zip archive, extracting the files, scanning the directory tree, parsing frontmatter schemas, and registering each skill.

In [1]python · cell 24

python

class Skill:
  def __init__(self, name, description, local_skill_dir):
    self.name = name
    self.description = description
    self.local_skill_dir = local_skill_dir

  def __str__(self):
    return f"Skill(name={self.name}, description={self.description}, local_skill_dir={self.local_skill_dir})"

def parse_skill_md(filepath: str) -> tuple[str, str]:
  """Parses SKILL.md file to get the skill name and description."""
  name = "Untitled Skill"
  description = "No description provided."
  try:
    with open(filepath, "r", encoding="utf-8") as f:
      content = f.read()
      if content.startswith("---"):
        end_idx = content.find("---", 3)
        if end_idx != -1:
          yaml_part = content[3:end_idx]
          try:
            data = yaml.safe_load(yaml_part)
            if data:
              name = data.get("name", name)
              description = data.get("description", description)
          except yaml.YAMLError as yaml_e:
            print(f"YAML parsing warning (using fallback): {yaml_e}")
            match_name = re.search(r"^name:\s*(.+)$", yaml_part, re.M)
            if match_name:
              name = match_name.group(1).strip()
            match_desc = re.search(
                r"^description:\s*(?:>-\s*)?(.+)$", yaml_part, re.M
            )
            if match_desc:
              description = match_desc.group(1).strip()
  except IOError as e:
    print(f"Failed to parse {filepath}: {e}")
  return name, description

def get_all_skills_from_dir(repo_dir: str) -> list[Skill]:
  """Returns a list of Skill objects from the given directory."""
  skills = []
  skills_found = 0
  seen_skills = set()
  for root, dirs, files in os.walk(repo_dir):
    lower_files = {f.lower(): f for f in files}
    if "skill.md" in lower_files:
      # Prune subdirectories to avoid deeper recursion in this skill folder.
      dirs[:] = []

      skill_md_filename = lower_files["skill.md"]
      filepath = os.path.join(root, skill_md_filename)

      name, description = parse_skill_md(filepath)

      # De-dupe based on skill name
      if name in seen_skills:
        print(f"Skipping duplicate skill name: {name}")
        continue
      seen_skills.add(name)

      skills.append(Skill(name, description, root))
      skills_found += 1
  print(f"Found {skills_found} skills.")
  return skills

def download_github_repo(repo_url, output_dir):
  os.makedirs(output_dir, exist_ok=True)
  branch = "master"

  zip_url = f"{GITHUB_REPO_URL}/archive/refs/heads/{branch}.zip"
  print(f"Attempting to download repository zip via HTTP: {zip_url}...")

  with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip_file:
    zip_path = temp_zip_file.name

  request = urllib.request.Request(zip_url)
  with urllib.request.urlopen(request, timeout=30) as response:
    with open(zip_path, "wb") as out_file:
      shutil.copyfileobj(response, out_file)
  with zipfile.ZipFile(zip_path, "r") as zip_ref:
    zip_ref.extractall(output_dir)
  print(
      f"Repository extracted successfully from '{branch}' branch zip"
      " archive."
  )
  os.remove(zip_path)
  print(f"Deleted temporary zip file: {zip_path}")

def create_skill(skill):
  name = skill.name
  description = skill.description
  skill_dir = skill.local_skill_dir
  print(f"Creating skill: {skill}")
  timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
  skill = client.skills.create(
      display_name=name,
      description=description,
      config={
          "local_path": skill_dir,
          "skill_id": f"{name}-{timestamp}"
      }
  )
  print(f"Created skill: {name}")

Run the batch ingestion pipeline

Specify your target repository URL, trigger the download, and execute the bulk skill ingestion. Each discovered skill will be individually registered with a unique timestamped identifier.

In [ ]python · cell 26

python

output_dir = tempfile.mkdtemp()
download_github_repo(GITHUB_REPO_URL, output_dir)

skills = get_all_skills_from_dir(output_dir)
for skill in skills:
    create_skill(skill)

Semantic Skill Discovery and Retrieval

The Skill Registry has built-in vector search capabilities. When skills are registered, their metadata descriptions are automatically vectorized and indexed.

This allows you to perform semantic search using natural language queries (e.g., searching for "firebase" or "database helpers") to locate the most relevant skills, even if there are no exact keyword matches.

Agents use this exact retrieval mechanism at runtime to dynamically discover and load matching skills based on the user's intent.

In [ ]python · cell 28

python

print(f"\n--- Search Relevant Skills ---")
query = "firebase"
print(f"Searching for skills matching query: '{query}'...")
try:
    # Call retrieve to perform semantic search
    response = client.skills.retrieve(
        query=query,
        config={"top_k": 2}
    )
    print(f"SUCCESS: Skills retrieved successfully!")
    print(f"Found {len(response.retrieved_skills)} matching skills.")
    for i, retrieved in enumerate(response.retrieved_skills):
        print(f"  [{i+1}] Skill Name: {retrieved.skill_name}")
        print(f"  [{i+1}] Description: {retrieved.description}")
except Exception as e:
    print(f"FAILED: Retrieve skills failed: {e}")

In [ ]python · cell 29

python

← 返回课程

Generative AI on Google Cloud

27 / 80·notebook

Chapter 27

Intro to Skill Registry

NotebookPython 329 cells

In [ ]python · cell 1

python

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Intro to Skill Registry

Authors
Shirong Bai
Deep Dubey
Eric Dong

Overview

The Skill Registry is a secure, private repository for storing, indexing, and dynamically discovering agent skills on the Gemini Enterprise Agent Platform.

Objectives

In this notebook, you will learn how to:

Define a Custom Skill: Create a local skill package structure with instructions and metadata.
Upload and Register a Skill: Programmatically register custom skills in the Skill Registry.
Ingest Skills from GitHub: Batch-import ready-to-use agent skills directly from a GitHub repository.
Perform Semantic Discovery: Retrieve and search relevant skills dynamically using semantic query matching.

Getting Started

Install libraries

In [ ]python · cell 7

python

%pip install --upgrade --quiet "google-cloud-aiplatform==1.152.0"

⚠️ Note: Ignore pip dependency errors.

Import libraries

In [ ]python · cell 10

python

import os
import re
import sys
import shutil
import tempfile
import urllib.request
import yaml
import zipfile
import vertexai

from datetime import datetime

Authenticate your notebook environment

If you are running this notebook in Google Colab, execute the cell below to authenticate.

In [ ]python · cell 12

python

if "google.colab" in sys.modules:
    from google.colab import auth

    auth.authenticate_user()

Autenticate your Google Cloud project

You can use a Google Cloud Project or an API Key for authentication. This tutorial uses a Google Cloud Project.

Enable the Agent Platform API

In [ ]python · cell 14

python

# fmt: off
PROJECT_ID = "[your-project-id]"  # @param {type: "string", placeholder: "[your-project-id]", isTemplate: true}
# fmt: on
if not PROJECT_ID or PROJECT_ID == "[your-project-id]":
    PROJECT_ID = str(os.getenv("GOOGLE_CLOUD_PROJECT"))

LOCATION = "us-central1"

Initialize the Agent Platform client

Initialize the SDK and instantiate the central platform client. This client is the entry point for managing custom resources (such as agents and skills) on the Gemini Enterprise Agent Platform.

In [ ]python · cell 16

python

vertexai.init(project=PROJECT_ID, location=LOCATION)

client = vertexai.Client(project=PROJECT_ID, location=LOCATION)

Define and Register a Custom Skill

To register a skill in the Skill Registry, you need a skill package. At its simplest, a skill package is a local directory containing a mandatory SKILL.md file.

The SKILL.md file follows a specific layout:

YAML Frontmatter: A metadata block at the top specifying:
- name: A unique identifier matching the skill package name.
- description: A capability statement (starting in third-person) explaining what the skill does.
Instructions: The core markdown instructions that guide the agent's behavior when executing the skill.

Create a sample skill package

The following cell programmatically creates a temporary directory (/tmp/sample_math_skill) containing a sample SKILL.md file for a simple math helper.

In [ ]python · cell 18

python

# Create a temporary directory to simulate a local skill package
local_skill_dir = "/tmp/sample_math_skill"
os.makedirs(local_skill_dir, exist_ok=True)

# Create the SKILL.md file which defines the skill
skill_md_content = """---
name: sample-math-skill
description: A sample skill that helps with simple math addition. Use this skill when you need to add two numbers.
---
# Test Math Skill
This skill provides instructions on how to add two numbers.

## Instructions
To add two numbers, sum them up. For example, 2 + 2 = 4.
"""

with open(os.path.join(local_skill_dir, "SKILL.md"), "w") as f:
    f.write(skill_md_content)

print(f"Created sample skill directory at: {local_skill_dir}")

Upload and register the skill package

Now, upload and register your custom skill in the Skill Registry using client.skills.create.

The SDK will automatically package, compress, and upload the local skill folder specified in "local_path".
By default, the operation blocks synchronously (wait_for_completion=True) until the backend has successfully provisioned and indexed the skill.

In [ ]python · cell 20

python

ts = datetime.now().strftime("%Y%m%d-%H%m%S")
user_skill_id = f"math-skill-{ts}"

try:
    # Call create with local_path in config
    # By default, wait_for_completion is True, so this will block until the skill is created
    skill = client.skills.create(
        display_name="Sample math skill",
        description="This skill provides functions to perform math calculations",
        config={
            "skill_id": user_skill_id,
            "local_path": "/tmp/sample_math_skill"
        }
        )
    print(f"SUCCESS: Skill created successfully!")
    print(f"Skill Name: {skill.name}")
    print(f"Skill State: {skill.state}")
    print(f"Skill Display Name: {skill.display_name}")
    print(f"Skill Description: {skill.description}")
except Exception as e:
    print(f"FAILED: Skill creation failed: {e}")
    skill = None

Batch Ingestion of Skills from a GitHub Repository

In real-world production environments, organization-wide agent skills are often stored and managed in a central Git repository.

This section demonstrates how to build an automated ingestion pipeline to fetch, parse, and batch-register skills from a remote repository.

We will use the Google-managed google/skills repository as our input. This repository contains ready-to-use agent skills, such as:

bigquery-basics: Fundamental tools and instructions for querying Google Cloud BigQuery datasets.
cloud-run-basics: Key operations and deployment guidelines for managing Google Cloud Run services.

In [ ]python · cell 22

python

GITHUB_REPO_URL = "https://github.com/google/skills" # @param {type:"string"}

Define pipeline helper functions

In [1]python · cell 24

python

class Skill:
  def __init__(self, name, description, local_skill_dir):
    self.name = name
    self.description = description
    self.local_skill_dir = local_skill_dir

  def __str__(self):
    return f"Skill(name={self.name}, description={self.description}, local_skill_dir={self.local_skill_dir})"

def parse_skill_md(filepath: str) -> tuple[str, str]:
  """Parses SKILL.md file to get the skill name and description."""
  name = "Untitled Skill"
  description = "No description provided."
  try:
    with open(filepath, "r", encoding="utf-8") as f:
      content = f.read()
      if content.startswith("---"):
        end_idx = content.find("---", 3)
        if end_idx != -1:
          yaml_part = content[3:end_idx]
          try:
            data = yaml.safe_load(yaml_part)
            if data:
              name = data.get("name", name)
              description = data.get("description", description)
          except yaml.YAMLError as yaml_e:
            print(f"YAML parsing warning (using fallback): {yaml_e}")
            match_name = re.search(r"^name:\s*(.+)$", yaml_part, re.M)
            if match_name:
              name = match_name.group(1).strip()
            match_desc = re.search(
                r"^description:\s*(?:>-\s*)?(.+)$", yaml_part, re.M
            )
            if match_desc:
              description = match_desc.group(1).strip()
  except IOError as e:
    print(f"Failed to parse {filepath}: {e}")
  return name, description

def get_all_skills_from_dir(repo_dir: str) -> list[Skill]:
  """Returns a list of Skill objects from the given directory."""
  skills = []
  skills_found = 0
  seen_skills = set()
  for root, dirs, files in os.walk(repo_dir):
    lower_files = {f.lower(): f for f in files}
    if "skill.md" in lower_files:
      # Prune subdirectories to avoid deeper recursion in this skill folder.
      dirs[:] = []

      skill_md_filename = lower_files["skill.md"]
      filepath = os.path.join(root, skill_md_filename)

      name, description = parse_skill_md(filepath)

      # De-dupe based on skill name
      if name in seen_skills:
        print(f"Skipping duplicate skill name: {name}")
        continue
      seen_skills.add(name)

      skills.append(Skill(name, description, root))
      skills_found += 1
  print(f"Found {skills_found} skills.")
  return skills

def download_github_repo(repo_url, output_dir):
  os.makedirs(output_dir, exist_ok=True)
  branch = "master"

  zip_url = f"{GITHUB_REPO_URL}/archive/refs/heads/{branch}.zip"
  print(f"Attempting to download repository zip via HTTP: {zip_url}...")

  with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip_file:
    zip_path = temp_zip_file.name

  request = urllib.request.Request(zip_url)
  with urllib.request.urlopen(request, timeout=30) as response:
    with open(zip_path, "wb") as out_file:
      shutil.copyfileobj(response, out_file)
  with zipfile.ZipFile(zip_path, "r") as zip_ref:
    zip_ref.extractall(output_dir)
  print(
      f"Repository extracted successfully from '{branch}' branch zip"
      " archive."
  )
  os.remove(zip_path)
  print(f"Deleted temporary zip file: {zip_path}")

def create_skill(skill):
  name = skill.name
  description = skill.description
  skill_dir = skill.local_skill_dir
  print(f"Creating skill: {skill}")
  timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
  skill = client.skills.create(
      display_name=name,
      description=description,
      config={
          "local_path": skill_dir,
          "skill_id": f"{name}-{timestamp}"
      }
  )
  print(f"Created skill: {name}")

Run the batch ingestion pipeline

Specify your target repository URL, trigger the download, and execute the bulk skill ingestion. Each discovered skill will be individually registered with a unique timestamped identifier.

In [ ]python · cell 26

python

output_dir = tempfile.mkdtemp()
download_github_repo(GITHUB_REPO_URL, output_dir)

skills = get_all_skills_from_dir(output_dir)
for skill in skills:
    create_skill(skill)

Semantic Skill Discovery and Retrieval

The Skill Registry has built-in vector search capabilities. When skills are registered, their metadata descriptions are automatically vectorized and indexed.

Agents use this exact retrieval mechanism at runtime to dynamically discover and load matching skills based on the user's intent.

In [ ]python · cell 28

python

print(f"\n--- Search Relevant Skills ---")
query = "firebase"
print(f"Searching for skills matching query: '{query}'...")
try:
    # Call retrieve to perform semantic search
    response = client.skills.retrieve(
        query=query,
        config={"top_k": 2}
    )
    print(f"SUCCESS: Skills retrieved successfully!")
    print(f"Found {len(response.retrieved_skills)} matching skills.")
    for i, retrieved in enumerate(response.retrieved_skills):
        print(f"  [{i+1}] Skill Name: {retrieved.skill_name}")
        print(f"  [{i+1}] Description: {retrieved.description}")
except Exception as e:
    print(f"FAILED: Retrieve skills failed: {e}")

In [ ]python · cell 29

python