文件 AWS 開發套件範例 GitHub 儲存庫中有更多可用的 [AWS SDK 範例](https://github.com/awsdocs/aws-doc-sdk-examples)。
本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 使用 SDK for Python (Boto3) 的 Step Functions 範例
下列程式碼範例示範如何使用 適用於 Python (Boto3) 的 AWS SDK 搭配 Step Functions 來執行動作和實作常見案例。
*基本概念*是程式碼範例，這些範例說明如何在服務內執行基本操作。
*Actions* 是大型程式的程式碼摘錄，必須在內容中執行。雖然動作會告訴您如何呼叫個別服務函數，但您可以在其相關情境中查看內容中的動作。
*案例*是向您展示如何呼叫服務中的多個函數或與其他 AWS 服務組合來完成特定任務的程式碼範例。
每個範例均包含完整原始碼的連結，您可在連結中找到如何設定和執行內容中程式碼的相關指示。
**Topics**
+ [開始使用](#get_started)
+ [基本概念](#basics)
+ [動作](#actions)
+ [案例](#scenarios)
## 開始使用
### Hello Step Functions
下列程式碼範例說明如何開始使用 Step Functions。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
import boto3
def hello_stepfunctions(stepfunctions_client):
"""
Use the AWS SDK for Python (Boto3) to create an AWS Step Functions client and list
the state machines in your account. This list might be empty if you haven't created
any state machines.
This example uses the default settings specified in your shared credentials
and config files.
:param stepfunctions_client: A Boto3 Step Functions Client object.
"""
print("Hello, Step Functions! Let's list up to 10 of your state machines:")
state_machines = stepfunctions_client.list_state_machines(maxResults=10)
for sm in state_machines["stateMachines"]:
print(f"\t{sm['name']}: {sm['stateMachineArn']}")
if __name__ == "__main__":
hello_stepfunctions(boto3.client("stepfunctions"))
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [ListStateMachines](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/ListStateMachines)。
## 基本概念
### 了解基本概念
以下程式碼範例顯示做法：
+ 建立活動。
+ 從 Amazon States Language 定義建立狀態機器，其中包含先前建立的活動步驟。
+ 執行狀態機器，並使用使用者輸入回應活動。
+ 在執行完成後取得最終狀態和輸出，然後清除資源。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
在命令提示中執行互動式案例。
```
class StateMachineScenario:
"""Runs an interactive scenario that shows how to get started using Step Functions."""
def __init__(self, activity, state_machine, iam_client):
"""
:param activity: An object that wraps activity actions.
:param state_machine: An object that wraps state machine actions.
:param iam_client: A Boto3 AWS Identity and Access Management (IAM) client.
"""
self.activity = activity
self.state_machine = state_machine
self.iam_client = iam_client
self.state_machine_role = None
def prerequisites(self, state_machine_role_name):
"""
Finds or creates an IAM role that can be assumed by Step Functions.
A role of this kind is required to create a state machine.
The state machine used in this example does not call any additional services,
so it needs no additional permissions.
:param state_machine_role_name: The name of the role.
:return: Data about the role.
"""
trust_policy = {
"Version":"2012-10-17",
"Statement": [
{
"Sid": "",
"Effect": "Allow",
"Principal": {"Service": "states.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
try:
role = self.iam_client.get_role(RoleName=state_machine_role_name)
print(f"Prerequisite IAM role {state_machine_role_name} already exists.")
except ClientError as err:
if err.response["Error"]["Code"] == "NoSuchEntity":
role = None
else:
logger.error(
"Couldn't get prerequisite IAM role %s. Here's why: %s: %s",
state_machine_role_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
if role is None:
try:
role = self.iam_client.create_role(
RoleName=state_machine_role_name,
AssumeRolePolicyDocument=json.dumps(trust_policy),
)
except ClientError as err:
logger.error(
"Couldn't create prerequisite IAM role %s. Here's why: %s: %s",
state_machine_role_name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
self.state_machine_role = role["Role"]
def find_or_create_activity(self, activity_name):
"""
Finds or creates a Step Functions activity.
:param activity_name: The name of the activity.
:return: The Amazon Resource Name (ARN) of the activity.
"""
print("First, let's set up an activity and state machine.")
activity_arn = self.activity.find(activity_name)
if activity_arn is None:
activity_arn = self.activity.create(activity_name)
print(
f"Activity {activity_name} created. Its Amazon Resource Name (ARN) is "
f"{activity_arn}."
)
else:
print(f"Activity {activity_name} already exists.")
return activity_arn
def find_or_create_state_machine(
self, state_machine_name, activity_arn, state_machine_file
):
"""
Finds or creates a Step Functions state machine.
:param state_machine_name: The name of the state machine.
:param activity_arn: The ARN of an activity that is used as a step in the state
machine. This ARN is injected into the state machine
definition that's used to create the state machine.
:param state_machine_file: The path to a file containing the state machine
definition.
:return: The ARN of the state machine.
"""
state_machine_arn = self.state_machine.find(state_machine_name)
if state_machine_arn is None:
with open(state_machine_file) as state_machine_file:
state_machine_def = state_machine_file.read().replace(
"{{DOC_EXAMPLE_ACTIVITY_ARN}}", activity_arn
)
state_machine_arn = self.state_machine.create(
state_machine_name,
state_machine_def,
self.state_machine_role["Arn"],
)
print(f"State machine {state_machine_name} created.")
else:
print(f"State machine {state_machine_name} already exists.")
print("-" * 88)
print(f"Here's some information about state machine {state_machine_name}:")
state_machine_info = self.state_machine.describe(state_machine_arn)
for field in ["name", "status", "stateMachineArn", "roleArn"]:
print(f"\t{field}: {state_machine_info[field]}")
return state_machine_arn
def run_state_machine(self, state_machine_arn, activity_arn):
"""
Run the state machine. The state machine used in this example is a simple
chat simulation. It contains an activity step in a loop that is used for user
interaction. When the state machine gets to the activity step, it waits for
an external application to get task data and submit a response. This function
acts as the activity application by getting task input and responding with
user input.
:param state_machine_arn: The ARN of the state machine.
:param activity_arn: The ARN of the activity used as a step in the state machine.
:return: The ARN of the run.
"""
print(
f"Let's run the state machine. It's a simplistic, non-AI chat simulator "
f"we'll call ChatSFN."
)
user_name = q.ask("What should ChatSFN call you? ", q.non_empty)
run_input = {"name": user_name}
print("Starting state machine...")
run_arn = self.state_machine.start(state_machine_arn, json.dumps(run_input))
action = None
while action != "done":
activity_task = self.activity.get_task(activity_arn)
task_input = json.loads(activity_task["input"])
print(f"ChatSFN: {task_input['message']}")
action = task_input["actions"][
q.choose("What now? ", task_input["actions"])
]
task_response = {"action": action}
self.activity.send_task_success(
activity_task["taskToken"], json.dumps(task_response)
)
return run_arn
def finish_state_machine_run(self, run_arn):
"""
Wait for the state machine run to finish, then print final status and output.
:param run_arn: The ARN of the run to retrieve.
"""
print(f"Let's get the final output from the state machine:")
status = "RUNNING"
while status == "RUNNING":
run_output = self.state_machine.describe_run(run_arn)
status = run_output["status"]
if status == "RUNNING":
print(
"The state machine is still running, let's wait for it to finish."
)
wait(1)
elif status == "SUCCEEDED":
print(f"ChatSFN: {json.loads(run_output['output'])['message']}")
else:
print(f"Run status: {status}.")
def cleanup(
self,
state_machine_name,
state_machine_arn,
activity_name,
activity_arn,
state_machine_role_name,
):
"""
Clean up resources created by this example.
:param state_machine_name: The name of the state machine.
:param state_machine_arn: The ARN of the state machine.
:param activity_name: The name of the activity.
:param activity_arn: The ARN of the activity.
:param state_machine_role_name: The name of the role used by the state machine.
"""
if q.ask(
"Do you want to delete the state machine, activity, and role created for this "
"example? (y/n) ",
q.is_yesno,
):
self.state_machine.delete(state_machine_arn)
print(f"Deleted state machine {state_machine_name}.")
self.activity.delete(activity_arn)
print(f"Deleted activity {activity_name}.")
self.iam_client.delete_role(RoleName=state_machine_role_name)
print(f"Deleted role {state_machine_role_name}.")
def run_scenario(self, activity_name, state_machine_name):
print("-" * 88)
print("Welcome to the AWS Step Functions state machines demo.")
print("-" * 88)
activity_arn = self.find_or_create_activity(activity_name)
state_machine_arn = self.find_or_create_state_machine(
state_machine_name,
activity_arn,
"../../../resources/sample_files/chat_sfn_state_machine.json",
)
print("-" * 88)
run_arn = self.run_state_machine(state_machine_arn, activity_arn)
print("-" * 88)
self.finish_state_machine_run(run_arn)
print("-" * 88)
self.cleanup(
state_machine_name,
state_machine_arn,
activity_name,
activity_arn,
self.state_machine_role["RoleName"],
)
print("-" * 88)
print("\nThanks for watching!")
print("-" * 88)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
try:
stepfunctions_client = boto3.client("stepfunctions")
iam_client = boto3.client("iam")
scenario = StateMachineScenario(
Activity(stepfunctions_client),
StateMachine(stepfunctions_client),
iam_client,
)
scenario.prerequisites("doc-example-state-machine-chat")
scenario.run_scenario("doc-example-activity", "doc-example-state-machine")
except Exception:
logging.exception("Something went wrong with the demo.")
```
定義包裝狀態機器動作的類別。
```
class StateMachine:
"""Encapsulates Step Functions state machine actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def create(self, name, definition, role_arn):
"""
Creates a state machine with the specific definition. The state machine assumes
the provided role before it starts a run.
:param name: The name to give the state machine.
:param definition: The Amazon States Language definition of the steps in the
the state machine.
:param role_arn: The Amazon Resource Name (ARN) of the role that is assumed by
Step Functions when the state machine is run.
:return: The ARN of the newly created state machine.
"""
try:
response = self.stepfunctions_client.create_state_machine(
name=name, definition=definition, roleArn=role_arn
)
except ClientError as err:
logger.error(
"Couldn't create state machine %s. Here's why: %s: %s",
name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response["stateMachineArn"]
def find(self, name):
"""
Find a state machine by name. This requires listing the state machines until
one is found with a matching name.
:param name: The name of the state machine to search for.
:return: The ARN of the state machine if found; otherwise, None.
"""
try:
paginator = self.stepfunctions_client.get_paginator("list_state_machines")
for page in paginator.paginate():
for state_machine in page.get("stateMachines", []):
if state_machine["name"] == name:
return state_machine["stateMachineArn"]
except ClientError as err:
logger.error(
"Couldn't list state machines. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
def describe(self, state_machine_arn):
"""
Get data about a state machine.
:param state_machine_arn: The ARN of the state machine to look up.
:return: The retrieved state machine data.
"""
try:
response = self.stepfunctions_client.describe_state_machine(
stateMachineArn=state_machine_arn
)
except ClientError as err:
logger.error(
"Couldn't describe state machine %s. Here's why: %s: %s",
state_machine_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
def start(self, state_machine_arn, run_input):
"""
Start a run of a state machine with a specified input. A run is also known
as an "execution" in Step Functions.
:param state_machine_arn: The ARN of the state machine to run.
:param run_input: The input to the state machine, in JSON format.
:return: The ARN of the run. This can be used to get information about the run,
including its current status and final output.
"""
try:
response = self.stepfunctions_client.start_execution(
stateMachineArn=state_machine_arn, input=run_input
)
except ClientError as err:
logger.error(
"Couldn't start state machine %s. Here's why: %s: %s",
state_machine_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response["executionArn"]
def describe_run(self, run_arn):
"""
Get data about a state machine run, such as its current status or final output.
:param run_arn: The ARN of the run to look up.
:return: The retrieved run data.
"""
try:
response = self.stepfunctions_client.describe_execution(
executionArn=run_arn
)
except ClientError as err:
logger.error(
"Couldn't describe run %s. Here's why: %s: %s",
run_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
def delete(self, state_machine_arn):
"""
Delete a state machine and all of its run data.
:param state_machine_arn: The ARN of the state machine to delete.
"""
try:
response = self.stepfunctions_client.delete_state_machine(
stateMachineArn=state_machine_arn
)
except ClientError as err:
logger.error(
"Couldn't delete state machine %s. Here's why: %s: %s",
state_machine_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
```
定義包裝活動動作的類別。
```
class Activity:
"""Encapsulates Step Function activity actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def create(self, name):
"""
Create an activity.
:param name: The name of the activity to create.
:return: The Amazon Resource Name (ARN) of the newly created activity.
"""
try:
response = self.stepfunctions_client.create_activity(name=name)
except ClientError as err:
logger.error(
"Couldn't create activity %s. Here's why: %s: %s",
name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response["activityArn"]
def find(self, name):
"""
Find an activity by name. This requires listing activities until one is found
with a matching name.
:param name: The name of the activity to search for.
:return: If found, the ARN of the activity; otherwise, None.
"""
try:
paginator = self.stepfunctions_client.get_paginator("list_activities")
for page in paginator.paginate():
for activity in page.get("activities", []):
if activity["name"] == name:
return activity["activityArn"]
except ClientError as err:
logger.error(
"Couldn't list activities. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
def get_task(self, activity_arn):
"""
Gets task data for an activity. When a state machine is waiting for the
specified activity, a response is returned with data from the state machine.
When a state machine is not waiting, this call blocks for 60 seconds.
:param activity_arn: The ARN of the activity to get task data for.
:return: The task data for the activity.
"""
try:
response = self.stepfunctions_client.get_activity_task(
activityArn=activity_arn
)
except ClientError as err:
logger.error(
"Couldn't get a task for activity %s. Here's why: %s: %s",
activity_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
def send_task_success(self, task_token, task_response):
"""
Sends a success response to a waiting activity step. A state machine with an
activity step waits for the activity to get task data and then respond with
either success or failure before it resumes processing.
:param task_token: The token associated with the task. This is included in the
response to the get_activity_task action and must be sent
without modification.
:param task_response: The response data from the activity. This data is
received and processed by the state machine.
"""
try:
self.stepfunctions_client.send_task_success(
taskToken=task_token, output=task_response
)
except ClientError as err:
logger.error(
"Couldn't send task success. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
def delete(self, activity_arn):
"""
Delete an activity.
:param activity_arn: The ARN of the activity to delete.
"""
try:
response = self.stepfunctions_client.delete_activity(
activityArn=activity_arn
)
except ClientError as err:
logger.error(
"Couldn't delete activity %s. Here's why: %s: %s",
activity_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
```
+ 如需 API 詳細資訊，請參閱《適用於 Python (Boto3) 的AWS SDK API 參考》**中的下列主題。
+ [CreateActivity](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/CreateActivity)
+ [CreateStateMachine](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/CreateStateMachine)
+ [DeleteActivity](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DeleteActivity)
+ [DeleteStateMachine](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DeleteStateMachine)
+ [DescribeExecution](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DescribeExecution)
+ [DescribeStateMachine](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DescribeStateMachine)
+ [GetActivityTask](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/GetActivityTask)
+ [ListActivities](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/ListActivities)
+ [ListStateMachines](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/ListStateMachines)
+ [SendTaskSuccess](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/SendTaskSuccess)
+ [StartExecution](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/StartExecution)
+ [StopExecution](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/StopExecution)
## 動作
### `CreateActivity`
以下程式碼範例顯示如何使用 `CreateActivity`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class Activity:
"""Encapsulates Step Function activity actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def create(self, name):
"""
Create an activity.
:param name: The name of the activity to create.
:return: The Amazon Resource Name (ARN) of the newly created activity.
"""
try:
response = self.stepfunctions_client.create_activity(name=name)
except ClientError as err:
logger.error(
"Couldn't create activity %s. Here's why: %s: %s",
name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response["activityArn"]
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [CreateActivity](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/CreateActivity)。
### `CreateStateMachine`
以下程式碼範例顯示如何使用 `CreateStateMachine`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class StateMachine:
"""Encapsulates Step Functions state machine actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def create(self, name, definition, role_arn):
"""
Creates a state machine with the specific definition. The state machine assumes
the provided role before it starts a run.
:param name: The name to give the state machine.
:param definition: The Amazon States Language definition of the steps in the
the state machine.
:param role_arn: The Amazon Resource Name (ARN) of the role that is assumed by
Step Functions when the state machine is run.
:return: The ARN of the newly created state machine.
"""
try:
response = self.stepfunctions_client.create_state_machine(
name=name, definition=definition, roleArn=role_arn
)
except ClientError as err:
logger.error(
"Couldn't create state machine %s. Here's why: %s: %s",
name,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response["stateMachineArn"]
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [CreateStateMachine](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/CreateStateMachine)。
### `DeleteActivity`
以下程式碼範例顯示如何使用 `DeleteActivity`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class Activity:
"""Encapsulates Step Function activity actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def delete(self, activity_arn):
"""
Delete an activity.
:param activity_arn: The ARN of the activity to delete.
"""
try:
response = self.stepfunctions_client.delete_activity(
activityArn=activity_arn
)
except ClientError as err:
logger.error(
"Couldn't delete activity %s. Here's why: %s: %s",
activity_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DeleteActivity](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DeleteActivity)。
### `DeleteStateMachine`
以下程式碼範例顯示如何使用 `DeleteStateMachine`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class StateMachine:
"""Encapsulates Step Functions state machine actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def delete(self, state_machine_arn):
"""
Delete a state machine and all of its run data.
:param state_machine_arn: The ARN of the state machine to delete.
"""
try:
response = self.stepfunctions_client.delete_state_machine(
stateMachineArn=state_machine_arn
)
except ClientError as err:
logger.error(
"Couldn't delete state machine %s. Here's why: %s: %s",
state_machine_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DeleteStateMachine](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DeleteStateMachine)。
### `DescribeExecution`
以下程式碼範例顯示如何使用 `DescribeExecution`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
def describe_run(self, run_arn):
"""
Get data about a state machine run, such as its current status or final output.
:param run_arn: The ARN of the run to look up.
:return: The retrieved run data.
"""
try:
response = self.stepfunctions_client.describe_execution(
executionArn=run_arn
)
except ClientError as err:
logger.error(
"Couldn't describe run %s. Here's why: %s: %s",
run_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DescribeExecution](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DescribeExecution)。
### `DescribeStateMachine`
以下程式碼範例顯示如何使用 `DescribeStateMachine`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class StateMachine:
"""Encapsulates Step Functions state machine actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def describe(self, state_machine_arn):
"""
Get data about a state machine.
:param state_machine_arn: The ARN of the state machine to look up.
:return: The retrieved state machine data.
"""
try:
response = self.stepfunctions_client.describe_state_machine(
stateMachineArn=state_machine_arn
)
except ClientError as err:
logger.error(
"Couldn't describe state machine %s. Here's why: %s: %s",
state_machine_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [DescribeStateMachine](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/DescribeStateMachine)。
### `GetActivityTask`
以下程式碼範例顯示如何使用 `GetActivityTask`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class Activity:
"""Encapsulates Step Function activity actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def get_task(self, activity_arn):
"""
Gets task data for an activity. When a state machine is waiting for the
specified activity, a response is returned with data from the state machine.
When a state machine is not waiting, this call blocks for 60 seconds.
:param activity_arn: The ARN of the activity to get task data for.
:return: The task data for the activity.
"""
try:
response = self.stepfunctions_client.get_activity_task(
activityArn=activity_arn
)
except ClientError as err:
logger.error(
"Couldn't get a task for activity %s. Here's why: %s: %s",
activity_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [GetActivityTask](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/GetActivityTask)。
### `ListActivities`
以下程式碼範例顯示如何使用 `ListActivities`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class Activity:
"""Encapsulates Step Function activity actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def find(self, name):
"""
Find an activity by name. This requires listing activities until one is found
with a matching name.
:param name: The name of the activity to search for.
:return: If found, the ARN of the activity; otherwise, None.
"""
try:
paginator = self.stepfunctions_client.get_paginator("list_activities")
for page in paginator.paginate():
for activity in page.get("activities", []):
if activity["name"] == name:
return activity["activityArn"]
except ClientError as err:
logger.error(
"Couldn't list activities. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [ListActivities](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/ListActivities)。
### `ListStateMachines`
以下程式碼範例顯示如何使用 `ListStateMachines`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
透過搜尋帳戶的狀態機器清單，依名稱尋找狀態機器。
```
class StateMachine:
"""Encapsulates Step Functions state machine actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def find(self, name):
"""
Find a state machine by name. This requires listing the state machines until
one is found with a matching name.
:param name: The name of the state machine to search for.
:return: The ARN of the state machine if found; otherwise, None.
"""
try:
paginator = self.stepfunctions_client.get_paginator("list_state_machines")
for page in paginator.paginate():
for state_machine in page.get("stateMachines", []):
if state_machine["name"] == name:
return state_machine["stateMachineArn"]
except ClientError as err:
logger.error(
"Couldn't list state machines. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [ListStateMachines](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/ListStateMachines)。
### `SendTaskSuccess`
以下程式碼範例顯示如何使用 `SendTaskSuccess`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class Activity:
"""Encapsulates Step Function activity actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def send_task_success(self, task_token, task_response):
"""
Sends a success response to a waiting activity step. A state machine with an
activity step waits for the activity to get task data and then respond with
either success or failure before it resumes processing.
:param task_token: The token associated with the task. This is included in the
response to the get_activity_task action and must be sent
without modification.
:param task_response: The response data from the activity. This data is
received and processed by the state machine.
"""
try:
self.stepfunctions_client.send_task_success(
taskToken=task_token, output=task_response
)
except ClientError as err:
logger.error(
"Couldn't send task success. Here's why: %s: %s",
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [SendTaskSuccess](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/SendTaskSuccess)。
### `StartExecution`
以下程式碼範例顯示如何使用 `StartExecution`。
**適用於 Python 的 SDK (Boto3)**
GitHub 上提供更多範例。尋找完整範例，並了解如何在 [AWS 程式碼範例儲存庫](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/stepfunctions#code-examples)中設定和執行。
```
class StateMachine:
"""Encapsulates Step Functions state machine actions."""
def __init__(self, stepfunctions_client):
"""
:param stepfunctions_client: A Boto3 Step Functions client.
"""
self.stepfunctions_client = stepfunctions_client
def start(self, state_machine_arn, run_input):
"""
Start a run of a state machine with a specified input. A run is also known
as an "execution" in Step Functions.
:param state_machine_arn: The ARN of the state machine to run.
:param run_input: The input to the state machine, in JSON format.
:return: The ARN of the run. This can be used to get information about the run,
including its current status and final output.
"""
try:
response = self.stepfunctions_client.start_execution(
stateMachineArn=state_machine_arn, input=run_input
)
except ClientError as err:
logger.error(
"Couldn't start state machine %s. Here's why: %s: %s",
state_machine_arn,
err.response["Error"]["Code"],
err.response["Error"]["Message"],
)
raise
else:
return response["executionArn"]
```
+ 如需 API 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [StartExecution](https://docs.aws.amazon.com/goto/boto3/states-2016-11-23/StartExecution)。
## 案例
### 建立傳訊應用程式
下列程式碼範例示範如何建立 AWS Step Functions 訊息應用程式，從資料庫資料表擷取訊息記錄。
**適用於 Python 的 SDK (Boto3)**
示範如何使用 適用於 Python (Boto3) 的 AWS SDK 搭配 AWS Step Functions 來建立訊息應用程式，從 Amazon DynamoDB 資料表擷取訊息記錄，並使用 Amazon Simple Queue Service (Amazon SQS) 傳送它們。狀態機器會與 AWS Lambda 函數整合，以掃描資料庫是否有未傳送的訊息。
+ 建立從 Amazon DynamoDB 資料表擷取和更新訊息記錄的狀態機器。
+ 更新狀態機器定義，以便也向 Amazon Simple Queue Service (Amazon SQS) 傳送訊息。
+ 開始和停用狀態機器執行。
+ 使用服務整合從狀態機器連接至 Lambda、DynamoDB 和 Amazon SQS。
如需完整的原始碼和如何設定及執行的指示，請參閱 [GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/stepfunctions_messenger) 上的完整範例。
**此範例中使用的服務**
+ DynamoDB
+ Lambda
+ Amazon SQS
+ 步驟函數
### 使用 Step Functions 協調生成式 AI 應用程式
下列程式碼範例示範如何使用 Amazon Bedrock 和 Step Functions，建置和協調生成式 AI 應用程式。
**適用於 Python 的 SDK (Boto3)**
Amazon Bedrock Serverless 提示鏈接案例展示，[AWS Step Functions](https://docs.aws.amazon.com/step-functions/latest/dg/welcome.html)、[Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/what-is-bedrock.html) 和 [https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html](https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html) 如何用於建置和協調複雜、無伺服器和可高度擴展的生成式 AI 應用程式。該案例包含下列工作範例：
+ 針對文獻部落格撰寫指定小說的分析。此範例描述簡易、循序的提示鏈。
+ 產生有關指定主題的簡短故事。此範例描述 AI 如何反覆處理先前產生的項目清單。
+ 建立前往指定目的地的週末假期行程。此範例描述如何平行處理多個不同的提示。
+ 向擔任電影製片的人類使用者推銷電影創意。此範例描述如何使用不同的推論參數平行處理相同的提示、如何回溯到鏈接的上一個步驟，以及如何將人工輸入包含在工作流程中。
+ 根據使用者手上的配料來規劃用餐。此範例描述提示鏈如何整合兩個不同的 AI 對話，其中兩個 AI 角色互相爭論以改善最終結果。
+ 尋找並總結目前最熱門的 GitHub 儲存庫。此範例說明鏈接多個與外部 API 互動的 AI 代理程式。
如需完整的原始碼，以及有關如何設定及執行的指示，請參閱 [GitHub](https://github.com/aws-samples/amazon-bedrock-serverless-prompt-chaining) 上的完整專案。
**此範例中使用的服務**
+ Amazon Bedrock
+ Amazon Bedrock 執行時期
+ Amazon Bedrock 代理程式
+ Amazon Bedrock 代理程式執行時期
+ 步驟函數