Setting the file. One moment.
Deploy Jumpstart Sagemaker · AWS AI ML · aws/agent-toolkit-for-aws · Skills Docs
Issue No. 14 · AWS AI ML
↖ Back to the coverMessaging And Streaming Skills
Migration And Modernization Skills
Networking And Content Delivery Skills
Security And Identity Skills
Web And Mobile Development
120 chapters · 648 min
Deploy Jumpstart Sagemaker Back to the top of the page 10
Setup DevOps Agent
24.7
Code Output Guide · references
RDS Oracle
(opens in a new tab)
references/model-deployment/code_templates/ deploy-jumpstart-sagemaker.py
Python · 177 lines · 7 KB
16 CONFIG = {
17 "model_id" : "[MODEL_ID]" , # REQUIRED — JumpStart model id, e.g. "huggingface-reasoning-qwen3-06b"
18 "model_version" : None , # None -> "*" (latest)
19 "instance_type" : "[INSTANCE_TYPE]" , # REQUIRED for a real-time endpoint
20 "instance_count" : 1 ,
21 "inference_config_name" : None , # a name from get_config_names, or None -> SDK top-ranked config
22 "accept_eula" : False , # True ONLY after explicit acceptance of a gated model's license
23 "env_vars" : None , # optional; merged over the config's defaults
24 }
25
26 # Runtime fields owned by deployment (NOT emitted by model-selection).
27 REGION = "[REGION]"
28 ROLE_ARN = "[ROLE_ARN]"
29 ENDPOINT_NAME = "[ENDPOINT_NAME]"
30 MODEL_NAME = "[MODEL_NAME]"
31 TAGS = [{ "Key" : "created-by" , "Value" : "sagemaker-agent-skill" }] # add user-provided tags too
32
33 os.environ[ "AWS_DEFAULT_REGION" ] = REGION
34
35
36 def validate_deployment_config (cfg):
37 """Fail fast with a clear, actionable message if the model-selection config is missing or
38 malformed. This guards the deployment against a broken hand-off — a bad config should stop
39 here with a readable error, not fail deep inside the SDK or (worse) silently misbehave.
40 """
41 if not isinstance (cfg, dict ):
42 raise ValueError (
43 f "Deployment config from model-selection must be a dict, got { type (cfg). __name__ } . "
44 "Re-run the model-selection skill to produce a valid config."
45 )
46 # model_id and instance_type are both required for a real-time endpoint. Reject empty values
47 # and un-substituted placeholders (e.g. "[MODEL_ID]").
48 for key in ( "model_id" , "instance_type" ):
49 val = cfg.get(key)
50 if not isinstance (val, str ) or not val.strip() or val.startswith( "[" ):
51 raise ValueError (
52 f "Deployment config is missing a valid ' { key } ' (got { val !r} ). "
53 "model-selection must emit it before deployment can proceed."
54 )
55 # instance_count must be a positive integer (bool is a subclass of int — reject it explicitly).
56 ic = cfg.get( "instance_count" , 1 )
57 if isinstance (ic, bool ) or not isinstance (ic, int ) or ic < 1 :
58 raise ValueError (
59 f "'instance_count' must be an integer >= 1, got { ic !r} . "
60 "model-selection must emit a valid instance_count."
61 )
62 # model_version is a non-empty string or None (None -> latest). An empty/whitespace string
63 # would reach JumpStartConfig(model_version="") and fail deep in the SDK.
64 mv = cfg.get( "model_version" )
65 if mv is not None and ( not isinstance (mv, str ) or not mv.strip()):
66 raise ValueError (
67 f "'model_version' must be a non-empty string or None (None -> latest), got { mv !r} ."
68 )
69 # inference_config_name is tri-state: a non-empty config name, or None for the SDK top-ranked
70 # config. An empty string would be treated as a real config name and fail deep in the SDK.
71 icn = cfg.get( "inference_config_name" )
72 if icn is not None and ( not isinstance (icn, str ) or not icn.strip()):
73 raise ValueError (
74 "'inference_config_name' must be a non-empty config name or None "
75 "(use None for the SDK's top-ranked config)."
76 )
77 # accept_eula MUST be a real bool. A non-bool such as the string "false" is truthy in Python
78 # and could silently auto-accept a gated model's license — a safety issue, so reject it.
79 eula = cfg.get( "accept_eula" , False )
80 if not isinstance (eula, bool ):
81 raise ValueError (
82 f "'accept_eula' must be a boolean, got { type (eula). __name__ } ( { eula !r} ). "
83 "A non-bool value could silently auto-accept a gated model's license."
84 )
85 # env_vars is None or a flat dict of string -> string.
86 ev = cfg.get( "env_vars" )
87 if ev is not None and (
88 not isinstance (ev, dict )
89 or not all ( isinstance (k, str ) and isinstance (v, str ) for k, v in ev.items())
90 ):
91 raise ValueError (
92 "'env_vars' must be None or a dict of string keys to string values. "
93 "model-selection must emit it in that shape."
94 )
95
96
97 validate_deployment_config( CONFIG )
98
99 # Cell 3: Build the model from the JumpStart config
100
101 # Cell 2 is intentionally SDK-free (stdlib only) so validate_deployment_config runs without the
102 # SageMaker SDK installed/importable; the SDK imports and set_attribution therefore live here in
103 # the build cell rather than the config cell.
104 from sagemaker.core import Attribution, set_attribution
105 from sagemaker.core.jumpstart.configs import JumpStartConfig
106 from sagemaker.core.training.configs import Compute
107 from sagemaker.serve import ModelBuilder
108
109 set_attribution(Attribution. SAGEMAKER_AGENT_PLUGIN )
110
111 # Map the validated config dict onto the two v3 SDK config objects. The SDK resolves the serving
112 # container, environment, and inference-component sizing internally from inference_config_name
113 # (None -> the spec's top-ranked config). accept_eula is a field on JumpStartConfig.
114 js = JumpStartConfig(
115 model_id = CONFIG [ "model_id" ],
116 model_version = CONFIG .get( "model_version" ), # None -> "*"
117 inference_config_name = CONFIG .get( "inference_config_name" ),
118 accept_eula = CONFIG .get( "accept_eula" , False ),
119 )
120 compute = Compute(
121 instance_type = CONFIG [ "instance_type" ],
122 instance_count = CONFIG .get( "instance_count" , 1 ),
123 )
124 mb = ModelBuilder.from_jumpstart_config(
125 jumpstart_config = js,
126 compute = compute,
127 role_arn = ROLE_ARN ,
128 env_vars = CONFIG .get( "env_vars" ) or None ,
129 )
130 model = mb.build( model_name = MODEL_NAME )
131 print ( f "Model built: { getattr (model, 'model_arn' , MODEL_NAME ) } " )
132
133 # Cell 4: Deploy to a real-time endpoint and wait for InService
134
135 import boto3
136
137 # from_jumpstart_config() deploys a single-model real-time endpoint. deploy() blocks until the
138 # endpoint is InService and returns a core Endpoint (use .invoke() on it).
139 endpoint = mb.deploy( endpoint_name = ENDPOINT_NAME )
140 print ( f "Deployed { CONFIG [ 'model_id' ] } to endpoint ' { ENDPOINT_NAME } '." )
141
142 # Tag for cost attribution / observability.
143 sm = boto3.client( "sagemaker" , region_name = REGION )
144 endpoint_arn = sm.describe_endpoint( EndpointName = ENDPOINT_NAME )[ "EndpointArn" ]
145 sm.add_tags( ResourceArn = endpoint_arn, Tags = TAGS )
146
147 # Cell 5: Test Inference
148
149 result = endpoint.invoke(
150 body = json.dumps(
151 { "inputs" : "What is the capital of France?" , "parameters" : { "max_new_tokens" : 50 }}
152 ),
153 content_type = "application/json" ,
154 )
155 print ( f "Response: { result.body.read().decode( 'utf-8' ) } " )
156
157 # Cell 6: Save Manifest
158 # Save manifest - record output of workflow step for future reference
159 from pathlib import Path
160
161 manifest_dir = Path( "[PROJECT_DIR]" ) / "manifests"
162 manifest_dir.mkdir( parents = True , exist_ok = True )
163 manifest_path = manifest_dir / f "deploy- { ENDPOINT_NAME } .json"
164 manifest_path.write_text(
165 json.dumps(
166 {
167 "endpoint_name" : ENDPOINT_NAME ,
168 "model_name" : MODEL_NAME ,
169 "model_id" : CONFIG [ "model_id" ],
170 "model_version" : CONFIG .get( "model_version" ),
171 "instance_type" : CONFIG [ "instance_type" ],
172 "inference_config_name" : CONFIG .get( "inference_config_name" ),
173 },
174 indent = 2 ,
175 )
176 )
177 print ( f "Manifest saved: { manifest_path } " )