env

job_variables

ANTHROPIC_API_KEY

from prefect import flow, task
import os
import anthropic
import backoff

@task(
    name="claude_message",
    retries=3,
    retry_delay_seconds=[5, 10, 30],
    timeout_seconds=120,
    tags=["anthropic"]  # optional: tie to a concurrency limit
)
def claude_message(user_prompt: str) -> str:
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

    @backoff.on_exception(backoff.expo, (anthropic.RateLimitError, anthropic.APIStatusError), max_time=60)
    def _call():
        resp = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=800,
            temperature=0,
            messages=[{"role": "user", "content": user_prompt}],
        )
        # Messages API returns content as a list of blocks
        parts = []
        for block in resp.content:
            if block.type == "text":
                parts.append(block.text)
        return "\n".join(parts)

    return _call()

@flow(name="llm-flow")
def llm_flow(prompt: str = "Summarize Prefect in 3 bullet points"):
    result = claude_message(prompt)
    return result

if __name__ == "__main__":
    llm_flow()

bedrock:InvokeModel

from prefect import flow, task
import json
import boto3
import backoff
import os

# Optionally set region via env var on the worker pod: AWS_REGION
BEDROCK_REGION = os.getenv("AWS_REGION", "us-east-1")
MODEL_ID = "anthropic.claude-3-5-sonnet-20241022-v1:0"

@task(
    name="claude_message_bedrock",
    retries=3,
    retry_delay_seconds=[5, 10, 30],
    timeout_seconds=120,
    tags=["anthropic"]  # optional: tie to a concurrency limit
)
def claude_message_bedrock(user_prompt: str) -> str:
    client = boto3.client("bedrock-runtime", region_name=BEDROCK_REGION)

    body = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 800,
        "temperature": 0,
        "messages": [{"role": "user", "content": [{"type": "text", "text": user_prompt}]}],
    }

    @backoff.on_exception(backoff.expo, Exception, max_time=60)
    def _call():
        resp = client.invoke_model(
            modelId=MODEL_ID,
            body=json.dumps(body),
            contentType="application/json",
            accept="application/json",
        )
        payload = json.loads(resp["body"].read().decode("utf-8"))
        parts = []
        for block in payload.get("output", {}).get("content", []):
            if block.get("type") == "text":
                parts.append(block.get("text", ""))
        return "\n".join(parts)

    return _call()

@flow(name="llm-flow-bedrock")
def llm_flow_bedrock(prompt: str = "Summarize Prefect in 3 bullet points"):
    result = claude_message_bedrock(prompt)
    return result

if __name__ == "__main__":
    llm_flow_bedrock()

prefect work-pool create "eks-pool" --type kubernetes