feat(agent_registry): support published_skills accessor and project alias

src/google/adk/integrations/agent_registry/__init__.py

@@ -13,7 +13,9 @@
 # limitations under the License.
 
 from .agent_registry import AgentRegistry
+from .agent_registry import PublishedSkills
 
 __all__ = [
     'AgentRegistry',
+    'PublishedSkills',
 ]

src/google/adk/integrations/agent_registry/agent_registry.py

@@ -33,6 +33,8 @@
 from google.adk.auth.auth_credential import AuthCredential
 from google.adk.auth.auth_schemes import AuthScheme
 from google.adk.integrations.agent_identity.gcp_auth_provider_scheme import GcpAuthProviderScheme
+from google.adk.skills import _utils
+from google.adk.skills.models import Skill
 from google.adk.telemetry.tracing import GCP_MCP_SERVER_DESTINATION_ID
 from google.adk.tools.base_tool import BaseTool
 from google.adk.tools.mcp_tool.mcp_session_manager import SseConnectionParams
@@ -172,6 +174,38 @@ def _is_google_api(url: str) -> bool:
   )
 
 
+_SKILL_RESOURCE_NAME_PATTERN = re.compile(
+    r"^projects/([^/]+)/locations/([^/]+)/skills/([^/]+)$"
+)
+
+
+class PublishedSkills:
+  """Accessor for interacting with published skills in Agent Registry."""
+
+  def __init__(self, registry: AgentRegistry):
+    self._registry = registry
+
+  def get(self, name: str) -> Skill:
+    """Retrieves and loads a published skill by full resource name.
+
+    Args:
+      name: Full resource name of the skill, in the format
+        ``projects/{project}/locations/{location}/skills/{skill_id}``.
+
+    Returns:
+      A loaded `Skill` ready to pass to `SkillToolset(skills=[...])`.
+
+    Raises:
+      ValueError: If the skill name does not match the expected resource name
+        format, or the skill does not contain a default revision.
+      RuntimeError: If an API request to fetch metadata or media fails.
+    """
+    return self._registry._fetch_published_skill_sync(name)
+
+
+_PublishedSkillsAccessor = PublishedSkills
+
+
 class AgentRegistry:
   """Client for interacting with the Google Cloud Agent Registry service.
 
@@ -189,20 +223,24 @@ def __init__(
       header_provider: (
           Callable[[ReadonlyContext], Dict[str, str]] | None
       ) = None,
+      project: str | None = None,
   ):
     """Initializes the AgentRegistry client.
 
     Args:
       project_id: The Google Cloud project ID.
       location: The Google Cloud location (region).
       header_provider: Optional provider for custom headers.
+      project: Optional alias for project_id.
     """
-    self.project_id = project_id
+    self.project_id = project_id or project
     self.location = location
 
     if not self.project_id or not self.location:
       raise ValueError("project_id and location must be provided")
 
+    self._published_skills = PublishedSkills(self)
+
     self._base_path = f"projects/{self.project_id}/locations/{self.location}"
     self._header_provider = header_provider
     try:
@@ -232,6 +270,14 @@ def __init__(
         else AGENT_REGISTRY_BASE_URL
     )
 
+  @property
+  def project(self) -> str | None:
+    return self.project_id
+
+  @property
+  def published_skills(self) -> PublishedSkills:
+    return self._published_skills
+
   def _get_auth_headers(self) -> Dict[str, str]:
     """Refreshes credentials and returns authorization headers."""
     try:
@@ -275,9 +321,12 @@ def _make_request(
       data: Dict[str, Any] = response.json()
       return data
     except requests.exceptions.HTTPError as e:
+      status_code = (
+          e.response.status_code if e.response is not None else "unknown"
+      )
+      error_text = e.response.text if e.response is not None else str(e)
       raise RuntimeError(
-          f"API request failed with status {e.response.status_code}:"
-          f" {e.response.text}"
+          f"API request failed with status {status_code}: {error_text}"
       ) from e
     except requests.exceptions.RequestException as e:
       raise RuntimeError(f"API request failed (network error): {e}") from e
@@ -689,6 +738,98 @@ def get_remote_a2a_agent(
     )
 
 
+  def get_published_skill(self, name: str) -> Skill:
+    """Retrieves and loads a published skill by full resource name.
+
+    Args:
+      name: Full resource name of the skill, in the format
+        ``projects/{project}/locations/{location}/skills/{skill_id}``.
+
+    Returns:
+      A loaded `Skill` ready to pass to `SkillToolset(skills=[...])`.
+    """
+    return self.published_skills.get(name)
+
+  def _download_media(
+      self,
+      path_or_url: str,
+      params: Dict[str, Any] | None = None,
+  ) -> bytes:
+    if path_or_url.startswith("http://") or path_or_url.startswith("https://"):
+      url = path_or_url
+    elif path_or_url.startswith("projects/"):
+      url = f"{self._base_url}/{path_or_url}"
+    else:
+      url = f"{self._base_url}/{self._base_path}/{path_or_url}"
+
+    quota_project_id = (
+        getattr(self._credentials, "quota_project_id", None) or self.project_id
+    )
+    headers = merge_tracking_headers(
+        {"x-goog-user-project": quota_project_id} if quota_project_id else {}
+    )
+    try:
+      response = self._session.get(
+          url,
+          headers=headers,
+          params=params,
+          allow_redirects=True,
+      )
+      if 300 <= response.status_code < 400 and (
+          "location" in response.headers or "Location" in response.headers
+      ):
+        redirect_url = response.headers.get("location") or response.headers.get(
+            "Location"
+        )
+        response = self._session.get(redirect_url, allow_redirects=True)
+      response.raise_for_status()
+      return bytes(response.content)
+    except requests.exceptions.HTTPError as e:
+      status_code = (
+          e.response.status_code if e.response is not None else "unknown"
+      )
+      error_text = e.response.text if e.response is not None else str(e)
+      raise RuntimeError(
+          f"API request failed with status {status_code}: {error_text}"
+      ) from e
+    except requests.exceptions.RequestException as e:
+      raise RuntimeError(f"API request failed (network error): {e}") from e
+    except Exception as e:
+      raise RuntimeError(f"API request failed: {e}") from e
+
+  def _fetch_published_skill_sync(self, name: str) -> Skill:
+
+    if not isinstance(name, str) or not _SKILL_RESOURCE_NAME_PATTERN.match(
+        name
+    ):
+      raise ValueError(
+          f"Invalid skill resource name '{name}'. Expected format: "
+          "'projects/{project}/locations/{location}/skills/{skill_id}'."
+      )
+
+    skill_data = self._make_request(name)
+    default_revision = skill_data.get("defaultRevision") or skill_data.get(
+        "default_revision"
+    )
+    if not default_revision:
+      raise ValueError(f"Skill '{name}' does not contain default revision.")
+
+    if default_revision.startswith("http://") or default_revision.startswith(
+        "https://"
+    ):
+      revision_url = default_revision
+    elif default_revision.startswith("projects/"):
+      revision_url = f"{self._base_url}/{default_revision}"
+    else:
+      clean_revision = default_revision.lstrip("/")
+      revision_url = f"{self._base_url}/{name}/{clean_revision}"
+
+    zip_bytes = self._download_media(revision_url, params={"alt": "media"})
+    skill = _utils._load_skill_from_zip_bytes(zip_bytes)
+    skill._uri = revision_url
+    return skill
+
+
 def _use_client_cert_effective() -> bool:
   """Returns whether client certificate should be used for mTLS."""
   try: