使用 SDK for Python (Boto3) 的步驟函數範例 - AWS SDK 程式碼範例

文件 AWS SDK AWS 範例 SDK 儲存庫中有更多可用的 GitHub 範例。

本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。

使用 SDK for Python (Boto3) 的步驟函數範例

下列程式碼範例示範如何使用 AWS SDK for Python (Boto3) 搭配 Step Functions 來執行動作和實作常見案例。

基本概念是程式碼範例,示範如何在服務中執行基本操作。

Actions 是大型程式的程式碼摘錄,必須在內容中執行。雖然 動作會示範如何呼叫個別服務函數,但您可以在其相關案例中查看內容中的動作。

案例是程式碼範例,示範如何透過呼叫服務內的多個函數或與其他函數結合來完成特定任務 AWS 服務。

每個範例都包含完整原始程式碼的連結,您可以在其中找到如何在內容中設定和執行程式碼的指示。

開始使用

下列程式碼範例示範如何開始使用 Step Functions。

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 ListStateMachines AWS SDK for Python (Boto3) Word 參考中的 API

基本概念

以下程式碼範例顯示做法:

  • 建立活動。

  • 從包含先前建立之活動作為步驟的 Amazon States 語言定義建立狀態機器。

  • 執行狀態機器並使用使用者輸入回應活動。

  • 在執行完成後取得最終狀態和輸出,然後清除資源。

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

在命令提示中執行互動式案例。

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

動作

下列程式碼範例示範如何使用 CreateActivity

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 CreateActivity AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 CreateStateMachine

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 CreateStateMachine AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 DeleteActivity

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 DeleteActivity AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 DeleteStateMachine

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 DeleteStateMachine AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 DescribeExecution

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 DescribeExecution AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 DescribeStateMachine

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 DescribeStateMachine AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 GetActivityTask

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 GetActivityTask AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 ListActivities

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 ListActivities AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 ListStateMachines

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

透過搜尋帳戶的狀態機器清單,依名稱尋找狀態機器。

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 詳細資訊,請參閱 ListStateMachines AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 SendTaskSuccess

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 SendTaskSuccess AWS SDK for Python (Boto3) Word 參考中的 API

下列程式碼範例示範如何使用 StartExecution

SDK for Python (Boto3)
注意

還有更多 on GitHub。尋找完整範例,並了解如何在 AWS 程式碼範例儲存庫中設定和執行。

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 詳細資訊,請參閱 StartExecution AWS SDK for Python (Boto3) Word 參考中的 API

案例

下列程式碼範例示範如何建立 AWS Step Functions 訊息應用程式,從資料庫資料表擷取訊息記錄。

SDK for Python (Boto3)

示範如何使用 AWS SDK for Python (Boto3) 搭配 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 上的完整範例。

此範例中使用的服務
  • DynamoDB

  • Lambda

  • Amazon SQS

  • Step Functions

下列程式碼範例示範如何使用 Amazon Bedrock 和 Step Functions 建置和協調生成的 AI 應用程式。

SDK for Python (Boto3)

Amazon Bedrock Serverless Prompt Chaining 案例示範 AWS Step FunctionsAmazon Bedrock 和 如何https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html用於建置和協調複雜、無伺服器和高度可擴展的生成 AI 應用程式。它包含下列工作範例:

  • 針對文獻部落格撰寫指定小說的分析。此範例說明簡單、循序的提示鏈。

  • 產生有關指定主題的簡短故事。此範例說明 AI 如何反覆處理先前產生的項目清單。

  • 建立前往指定目的地的週末假期行程。此範例說明如何平行處理多個不同的提示。

  • 將電影想法調適給擔任電影製作者的人類使用者。此範例說明如何使用不同的推論參數平行處理相同的提示,如何恢復到鏈中的上一個步驟,以及如何將人工輸入納入工作流程中。

  • 根據使用者手頭的成分來規劃餐點。此範例說明提示鏈如何整合兩個不同的 AI 對話,其中兩個 AI 角色彼此進行爭論以改善最終結果。

  • 尋找並摘要今天最熱門的 GitHub 儲存庫。此範例說明連結多個與外部 APIs 互動的 AI 代理程式。

如需設定和執行的完整原始程式碼和指示,請參閱 GitHub 上的完整專案。

此範例中使用的服務
  • Amazon Bedrock

  • Amazon Bedrock 執行期

  • Amazon Bedrock 代理程式

  • Amazon Bedrock Agents 執行期

  • Step Functions