Setting the file. One moment.
Fetch Skill · Dd Orchestrator · datadog-labs/agent-skills · Skills Docs
Back to the top of the page Script
scripts/ fetch_skill.py
Python · 131 lines · 6 KB
15 """
16 from __future__ import annotations
17
18 import argparse
19 import json
20 import os
21 import sys
22 import tempfile
23 import urllib.request
24 from urllib.parse import urlsplit
25
26 HERE = os.path.dirname(os.path.abspath( __file__ ))
27 SKILL = os.path.dirname( HERE )
28 CATALOG = os.path.join( SKILL , "catalog.json" )
29
30 RAW_HOST = "raw.githubusercontent.com"
31 ALLOWED_GH_REPO = "datadog-labs/agent-skills" # the only public GitHub source
32 API_HOST = "api.datadoghq.com" # dd-source onboarding render
33 BRANCH = "main" # always newest; never pinned
34
35
36 def _get (url, binary = False ):
37 req = urllib.request.Request(url, headers = { "User-Agent" : "dd-orchestrator-fetch" })
38 with urllib.request.urlopen(req, timeout = 20 ) as resp:
39 if resp.status != 200 :
40 raise RuntimeError ( f "HTTP { resp.status } for { url } " )
41 data = resp.read()
42 return data if binary else data.decode( "utf-8" )
43
44
45 def resolve_target (node):
46 """Pure (no network): the newest-version fetch target for a catalog node.
47
48 Returns ("github", "datadog-labs/agent-skills", "<skill_dir>") — fetch dir at main
49 or ("render", "<onboarding-api render url>") — single md, newest
50 Raises for any non-public / unknown source (fail closed).
51 """
52 src = node.get( "source" ) or {}
53 url = src.get( "url" , "" )
54 host = urlsplit(url).netloc
55 if src.get( "repo" ) == "agent-skills" and host == "github.com" :
56 parts = urlsplit(url).path.strip( "/" ).split( "/" ) # org/repo/blob/<ref>/<path...>
57 if len (parts) < 5 or parts[ 2 ] not in ( "blob" , "tree" ):
58 raise RuntimeError ( f "malformed agent-skills url: { url !r} " )
59 org_repo = f " { parts[ 0 ] } / { parts[ 1 ] } "
60 if org_repo != ALLOWED_GH_REPO :
61 raise RuntimeError ( f "refusing non-allowlisted repo: { org_repo } " )
62 rel = "/" .join(parts[ 4 :])
63 skill_dir = os.path.dirname(rel) if src.get( "path_type" , "file" ) == "file" else rel
64 return ( "github" , org_repo, skill_dir)
65 if host == API_HOST and "/onboarding/skills/" in url:
66 return ( "render" , url)
67 raise RuntimeError ( f "unsupported/non-public source: { url !r} " )
68
69
70 def _fetch_github_dir (org_repo, dir_path, dest):
71 """Recursively fetch a skill directory from agent-skills @ main (contents API + raw)."""
72 api = f "https://api.github.com/repos/ { org_repo } /contents/ { dir_path } ?ref= { BRANCH } "
73 for entry in json.loads(_get(api)):
74 if entry[ "type" ] == "dir" :
75 _fetch_github_dir(org_repo, f " { dir_path } / { entry[ 'name' ] } " ,
76 os.path.join(dest, entry[ "name" ]))
77 elif entry[ "type" ] == "file" :
78 raw = f "https:// { RAW_HOST } / { org_repo } / { BRANCH } / { dir_path } / { entry[ 'name' ] } "
79 os.makedirs(dest, exist_ok = True )
80 with open (os.path.join(dest, entry[ "name" ]), "wb" ) as f:
81 f.write(_get(raw, binary = True ))
82
83
84 def _node (catalog_path, skill_id):
85 with open (catalog_path) as f:
86 for n in json.load(f)[ "nodes" ]:
87 if n[ "id" ] == skill_id:
88 return n
89 raise RuntimeError ( f "skill id { skill_id !r} not in catalog { catalog_path } " )
90
91
92 def fetch (skill_id, dest = None , catalog_path = CATALOG ):
93 node = _node(catalog_path, skill_id)
94 kind, * rest = resolve_target(node)
95 # Run-scoped destination: a fresh dir per fetch so a partial/older fetch can never leave
96 # stale files (e.g. an upstream-deleted script or a leftover SKILL.md) to be executed, and
97 # concurrent fetches cannot interleave. Caller reads the returned path, so a unique dir is fine.
98 dest = dest or tempfile.mkdtemp( prefix = f "dd-orch- { skill_id } -" )
99 os.makedirs(dest, exist_ok = True )
100 if kind == "github" :
101 org_repo, skill_dir = rest
102 _fetch_github_dir(org_repo, skill_dir, dest)
103 else : # render
104 (url,) = rest
105 with open (os.path.join(dest, "SKILL.md" ), "w" ) as f:
106 f.write(_get(url))
107 if not os.path.exists(os.path.join(dest, "SKILL.md" )):
108 raise RuntimeError ( f "fetched { skill_id } but no SKILL.md landed in { dest } " )
109 return dest
110
111
112 def main (argv = None ):
113 ap = argparse.ArgumentParser( description = "fetch a catalog skill from its public source (newest)" )
114 ap.add_argument( "skill_id" )
115 ap.add_argument( "--dest" )
116 ap.add_argument( "--catalog" , default = CATALOG )
117 ap.add_argument( "--plan" , action = "store_true" , help = "print the resolved target; do not fetch" )
118 args = ap.parse_args(argv)
119 try :
120 if args.plan:
121 print (resolve_target(_node(args.catalog, args.skill_id)))
122 else :
123 print (fetch(args.skill_id, args.dest, args.catalog))
124 except Exception as exc: # fail closed with a clear reason + non-zero exit
125 print ( f "fetch_skill: { exc } " , file = sys.stderr)
126 return 1
127 return 0
128
129
130 if __name__ == "__main__" :
131 raise SystemExit (main())