Skip to content

SageMakerAIModel

Invoke a deployed SageMaker endpoint. Requires the sagemaker extra and AWS credentials:

bash
pip install --extra-index-url https://core-packages.elsai.ai/root/ "elsai-model[sagemaker]==2.1.0"

Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION, SAGEMAKER_ENDPOINT_NAME

python
import os
from elsai_model.sagemaker import SageMakerAIModel

model = SageMakerAIModel(
    endpoint_config={
        "endpoint_name": os.environ["SAGEMAKER_ENDPOINT_NAME"],
        "region_name": os.getenv("AWS_REGION", "us-west-2"),
    },
    payload_config={"max_tokens": 256, "temperature": 0.2, "stream": False},
)
messages = [{"role": "user", "content": "Say hello in one short sentence."}]

response = model.invoke(messages)
print(response["choices"][0]["message"]["content"])

Or via the factory (no model=; use endpoint + payload config):

python
from elsai_model import LLM, Provider

model = LLM(
    provider=Provider.SAGEMAKER,
    endpoint_config={
        "endpoint_name": os.environ["SAGEMAKER_ENDPOINT_NAME"],
        "region_name": os.getenv("AWS_REGION", "us-west-2"),
    },
    payload_config={"max_tokens": 256, "temperature": 0.2, "stream": False},
)

Streaming uses the async stream() API (stream_text() is not available for SageMaker):

python
import asyncio
import os
from elsai_model.sagemaker import SageMakerAIModel

async def stream_response() -> None:
    model = SageMakerAIModel(
        endpoint_config={
            "endpoint_name": os.environ["SAGEMAKER_ENDPOINT_NAME"],
            "region_name": os.getenv("AWS_REGION", "us-west-2"),
        },
        payload_config={"max_tokens": 256, "temperature": 0.2, "stream": True},
    )
    messages = [{"role": "user", "content": [{"text": "Count from 1 to 5, one number per line."}]}]

    async for event in model.stream(messages):
        if "contentBlockDelta" in event:
            delta = event["contentBlockDelta"].get("delta", {})
            if "text" in delta:
                print(delta["text"], end="", flush=True)
    print()

asyncio.run(stream_response())

See also

Copyright © 2026 elsai foundry.