ワークフローを実行してレスポンスを取得する方法を学習します。
Workflow.run() 関数はエージェントを実行して、WorkflowRunOutput オブジェクトまたは WorkflowRunOutput オブジェクトのストリームとしてレスポンスを生成します。
Agno 2.x : Learn : ワークフロー – ワークフローの実行
作成 : クラスキャット・セールスインフォメーション
作成日時 : 11/12/2025
バージョン : Agno 2.2.10
* 本記事は docs.agno.com の以下のページを独自に翻訳した上で、補足説明を加えてまとめ直しています。スニペットはできる限り日本語を使用しています :
* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。
Agno 2.x : Learn : ワークフロー – ワークフローの実行
ワークフローを実行してレスポンスを取得する方法を学習します。
Workflow.run() 関数はエージェントを実行して、WorkflowRunOutput オブジェクトまたは WorkflowRunOutput オブジェクトのストリームとしてレスポンスを生成します。
多くのサンプルは workflow.print_response() を使用しています、これは端末でレスポンスを表示するためのヘルパーユティリティです。これは内部的には workflow.run() を使用しています。
ワークフローの実行
ワークフローを実行する方法は以下のとおりです。応答は response で取得できます。
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb
from agno.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.hackernews import HackerNewsTools
from agno.workflow import Step, Workflow
from agno.run.workflow import WorkflowRunOutput
from agno.utils.pprint import pprint_run_response
# Define agents
hackernews_agent = Agent(
name="Hackernews Agent",
model=OpenAIChat(id="gpt-5-mini"),
tools=[HackerNewsTools()],
role="Extract key insights and content from Hackernews posts",
)
web_agent = Agent(
name="Web Agent",
model=OpenAIChat(id="gpt-5-mini"),
tools=[DuckDuckGoTools()],
role="Search the web for the latest news and trends",
)
# Define research team for complex analysis
research_team = Team(
name="Research Team",
members=[hackernews_agent, web_agent],
instructions="Research tech topics from Hackernews and the web",
)
content_planner = Agent(
name="Content Planner",
model=OpenAIChat(id="gpt-5-mini"),
instructions=[
"Plan a content schedule over 4 weeks for the provided topic and research content",
"Ensure that I have posts for 3 posts per week",
],
)
content_creation_workflow = Workflow(
name="Content Creation Workflow",
description="Automated content creation from blog posts to social media",
db=SqliteDb(db_file="tmp/workflow.db"),
steps=[research_team, content_planner],
)
# Create and use workflow
if __name__ == "__main__":
response: WorkflowRunOutput = content_creation_workflow.run(
input="AI trends in 2024",
markdown=True,
)
pprint_run_response(response, markdown=True)
出力例
INFO Setting default model to OpenAI Chat ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ 以下は、いただいたAIトレンド情報をもとにした4週間のコンテンツスケジュール案です。週3回の投稿を前提に、全12投稿を計画しています。主要テーマを週… │ │ │ │ 前提 │ │ │ │ • 想定フォーマット例: │ │ LinkedIn記事（主投稿）、Twitter/Xスレッドの補完、ブログ／メディア用長文記事、スライドデッキ用のアウトライン、簡易要約リストなどを組み合わせ │ │ • 主な対象読者: 企業の意思決定者、プロダクト/エンジニアリングリーダー、マーケティング・事業部門、ヘルスケア部門の実務担当者 │ │ • 使用キーワード・ハッシュタグ例: #AI2025 #SpatialAI #AIUX #LegacyAI #HealthcareAI #AIMarketing #CryptoIPO #DigitalTrust #GartnerTrends │ │ │ │ Week 1: トレンドの基礎と実務の入口 │ │ │ │ • 投稿1（Mon）Gitの使い勝手向上とAI開発のUX │ │ • タイトル案: AI開発を速くするGit UXの未来 │ │ • 内容の要点: │ │ • AIプロジェクトでのGit UXが開発速度とチームコラボに与える影響 │ │ • 想定されるUX改善点（差分表示・AIアシストによるコードレビュ―・ワークフローのテンプレ化など） │ │ • 実務で取り入れられる小さな改善ステップの紹介 │ │ • 形式: LinkedIn記事（約800–1,200語）＋短いXスレッド │ │ • CTA: 「あなたのチームで検討しているGitUXの改善は何ですか？コメントで教えてください」 │ │ • ハッシュタグ: #AIUX #GitUX #AI2025 │ │ • 出典・補足: HackernewsのUX改善話題をベースに、実務のワークフロー改善観点を付加 │ │ • 投稿2（Wed）空間的知能（スペーシャルインテリジェンス） │ │ • タイトル案: 空間的知能が切り開くAIの新しい世界認識 │ │ • 内容の要点: │ │ • 空間的推論・世界認識の重要性と今後の適用領域（ロボティクス、AR/VR、センサー統合など） │ │ • 代表的なアーキテクチャの概観と、実務での適用可能性の解説 │ │ • 倫理・セーフティの観点とリスク │ │ • 形式: ブログ形式の長文記事（約1,000–1,500語） │ │ • CTA: 「実務で空間的知能をどう検討しますか？実例を教えてください」 │ │ • ハッシュタグ: #SpatialAI #AI2025 #WorldModel │ │ • 出典・補足: Hackernewsの「空間的知能」トピックを軸に、事例想定を追加 │ │ • 投稿3（Fri）レガシーシステムのAI利用（COBOL/メインフレーム） │ │ • タイトル案: レガシーとAIの共生: COBOL・メインフレームをどう活用するか │ │ • 内容の要点: │ │ • レガシーシステムの価値をAIで最大化する現実的アプローチ（AIラッパー、データ連携、レガシーから新セグメントへ移行する戦略） │ │ • 実務例のカテゴリ（監視・最適化・自動化のグラデーション） │ │ • 取り組み開始のロードマップ（初期ハードルと成功指標） │ │ • 形式: LinkedIn記事（約1,000語程度）と補足スライド案 │ │ • CTA: 「自社のレガシー環境でAIをどう活用しますか？アイデアを共有ください」 │ │ • ハッシュタグ: #LegacyAI #Mainframe #AI2025 │ │ • 出典・補足: トピックを実務視点で解説、導入の第一歩を具体化 │ │ │ │ Week 2: 実務適用と信頼・ヘルスケア・スタートアップ │ │ │ │ • 投稿1（Mon）実務適用とデジタル信頼の強化 │ │ • タイトル案: 実務適用のAIとデジタル信頼の新しい指針 │ │ • 内容の要点: │ │ • 透明性・信頼性の重要性と企業価値への影響 │ │ • データ・モデルのプロパエンテキセンス、ガバナンス、説明責任の枠組み │ │ • Gartnerの2026年トレンドの方向性との整合性 │ │ • 形式: LinkedIn記事（約1,000–1,300語）＋短い要約リスト │ │ • CTA: 「自社のAIガバナンスで最も課題だと感じる点は何ですか？」 │ │ • ハッシュタグ: #DigitalTrust #AITransparency #GartnerTrends │ │ • 投稿2（Wed）医療・ヘルスケアAI戦略 │ │ • タイトル案: 医療AI戦略の現場実装: 臨床支援と画像診断の効率化 │ │ • 内容の要点: │ │ • 臨床支援・画像診断の実務での適用例と導入ポイント │ │ • データ品質・臨床ワークフローとの統合 │ │ • 導入のリスク管理と規制対応の観点 │ │ • 形式: ブログ記事（約1,000–1,400語） │ │ • CTA: 「貴院・貴社の医療AI導入での課題は何ですか？共有できるベストプラクティスを教えてください」 │ │ • ハッシュタグ: #HealthcareAI #ClinicalSupport #AIInMedicine │ │ • 投稿3（Fri）AIスタートアップの台頭 │ │ • タイトル案: AIスタートアップの潮流と企業の活用機会 │ │ • 内容の要点: │ │ • ニッチ領域での特化型AIの採用が進む理由 │ │ • 各セグメントの事例とパートナーシップの在り方 │ │ • 企業が大企業としてどう接点を作るべきか（共同開発・Poc・検証フェーズの進め方） │ │ • 形式: LinkedIn記事＋リスト形式のまとめ │ │ • CTA: 「自社に適したAIスタートアップの見極め方は？」 │ │ • ハッシュタグ: #AIEcosystem #AIStartups #TechTrends │ │ • 出典・補足: 市場動向と実務適用の橋渡しを意識 │ │ │ │ Week 3: マーケティング・金融市場の動向と実践ロードマップ │ │ │ │ • 投稿1（Mon）マーケティングでのAI活用 │ │ • タイトル案: マーケティングの新潮流: AIが切り開くインフルエンサーマーケと広告技術 │ │ • 内容の要点: │ │ • インフルエンサーマーケティング・広告技術におけるAIの役割 │ │ • 実務の成功事例・失敗事例の教訓 │ │ • リスク管理と倫理的配慮 │ │ • 形式: LinkedIn記事（約1,000–1,200語）＋短いスレッド │ │ • CTA: 「あなたのブランドでAIマーケをどう実装しますか？」 │ │ • ハッシュタグ: #AIMarketing #AdTech #InfluencerTech │ │ • 投稿2（Wed）AIと暗号技術のIPO動向 │ │ • タイトル案: AIと暗号のIPO急増が示す2025年市場の新潮流 │ │ • 内容の要点: │ │ • 2025年のIPO市場の背景とAI・暗号分野の動向 │ │ • 投資家観点・リスク管理のポイント │ │ • 企業が注意すべき規制・ガバナンスの観点 │ │ • 形式: ブログ／長文記事（約1,200–1,500語） │ │ • CTA: 「市場の動きを見据えた企業戦略は？」 │ │ • ハッシュタグ: #CryptoIPO #AIMarket #FinanceTech │ │ • 投稿3（Fri）研究と実務を結ぶロードマップ │ │ • タイトル案: 研究成果をビジネスに落とす4つのステップ │ │ • 内容の要点: │ │ • 研究と実務のギャップを埋めるための具体的手順 │ │ • 小規模実証（Poc）→パイロット→スケールの道筋 │ │ • 成功指標・組織のガバナンス設計 │ │ • 形式: LinkedIn記事＋実行可能なチェックリスト │ │ • CTA: 「自社の研究開発を実務へどのように結びつけますか？」 │ │ • ハッシュタグ: #AIInPractice #InnovationRoadmap │ │ │ │ Week 4: 総括と実践のためのアクションプラン │ │ │ │ • 投稿1（Mon）2025年の総括と統合動向 │ │ • タイトル案: 2025年を席巻するトレンドの統合像 │ │ • 内容の要点: │ │ • 多言語ASR（約1600言語対応）・空間知能・Git UXの統合的展望 │ │ • アーキテクチャパターンの傾向と組織適応戦略 │ │ • 形式: LinkedIn長文記事（約1,000–1,400語）＋要点リスト │ │ • CTA: 「来年までに取り組むべき4つの優先事項を教えてください」 │ │ • ハッシュタグ: #AI2025 #FutureOfAI │ │ • 投稿2（Wed）責任あるAIの運用設計 │ │ • タイトル案: 責任あるAIを組織に組み込む設計指針 │ │ • 内容の要点: │ │ • ガバナンス、セキュリティ、透明性の4つの柱 │ │ • 実務での適用フレームとチェックリスト │ │ • 形式: ブログ記事（約1,000–1,300語）＋簡易ガイドライン │ │ • CTA: 「あなたの組織のAIガバナンス点検リストを作成しますか？」 │ │ • ハッシュタグ: #DigitalTrust #AIGovernance │ │ • 投稿3（Fri）実践用の90日ロードマップとリソース案内 │ │ • タイトル案: AI実装を成功させる90日計画 │ │ • 内容の要点: │ │ • 初期設定・データ readiness・パイロット設計・評価指標の設定 │ │ • リソース、ツール、学習リソースの優先順位 │ │ • 成功体験の共有と継続的な改善の仕組み │ │ • 形式: LinkedIn記事＋チェックリスト形式のリソース案内 │ │ • CTA: 「あなたの組織向けの90日ロードマップを一緒に作成します」 │ │ • ハッシュタグ: #AIImplementation #90DayPlan │ │ │ │ 補足と実行のコツ │ │ │ │ • 各週の3投稿は、相互に補完するように設計しています。例えばWeek 1のGitUXとSpatialAIをWeek │ │ 2のデジタル信頼や医療AI戦略と組み合わせることで、読者の理解を段階的に深められます。 │ │ • 投稿フォーマットの組み合わせ例 │ │ • 主投稿: LinkedIn記事または会社ブログ │ │ • 補足: 同内容のXスレッド、要点リスト、スライドデッキ（ピッチ用） │ │ • クロスリファレンス: 各投稿の末尾に「次回は〇〇を取り上げます」と予告して継続性を維持 │ │ • 写真・図表の活用 │ │ • 空間的知能の世界モデル図、レガシーAIのアーキテクチャ比較、デジタル信頼のガバナンスフレームの図表などを用意すると理解が深まります。 │ │ • 参考リサーチの整理 │ │ • Hackernewsのトピック（Git UX、空間的知能、レガシーシステムのAI利用、多言語ASR） │ │ • ウェブ情報（デジタル信頼、医療AI、AIスタートアップ、マーケティングAI、AIと暗号のIPO） │ │ • 上記トピックをベースに、実務事例・実用ガイド・ロードマップの形で分かりやすく解説する構成を推奨します。 │ │ │ │ もし、特定のトピックをさらに深掘りしたい場合や、予定日の日付を確定したい場合は教えてください。4週間のスケジュールをあなたの公開プラットフォー … │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
❗️ The Workflow.run() function returns a WorkflowRunOutput object when not streaming. Here is detailed documentation for WorkflowRunOutput.
非同期実行
Workflow.arun() 関数は Workflow.run() の非同期バージョンです。その使用方法の例は以下です :
from typing import AsyncIterator
import asyncio
from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.workflow import Condition, Step, Workflow, StepInput
from agno.run.workflow import WorkflowRunOutput, WorkflowRunOutputEvent, WorkflowRunEvent
# === BASIC AGENTS ===
researcher = Agent(
name="Researcher",
instructions="Research the given topic and provide detailed findings.",
tools=[DuckDuckGoTools()],
)
summarizer = Agent(
name="Summarizer",
instructions="Create a clear summary of the research findings.",
)
fact_checker = Agent(
name="Fact Checker",
instructions="Verify facts and check for accuracy in the research.",
tools=[DuckDuckGoTools()],
)
writer = Agent(
name="Writer",
instructions="Write a comprehensive article based on all available research and verification.",
)
# === CONDITION EVALUATOR ===
def needs_fact_checking(step_input: StepInput) -> bool:
"""Determine if the research contains claims that need fact-checking"""
summary = step_input.previous_step_content or ""
# Look for keywords that suggest factual claims
fact_indicators = [
"study shows",
"breakthroughs",
"research indicates",
"according to",
"statistics",
"data shows",
"survey",
"report",
"million",
"billion",
"percent",
"%",
"increase",
"decrease",
]
return any(indicator in summary.lower() for indicator in fact_indicators)
# === WORKFLOW STEPS ===
research_step = Step(
name="research",
description="Research the topic",
agent=researcher,
)
summarize_step = Step(
name="summarize",
description="Summarize research findings",
agent=summarizer,
)
# Conditional fact-checking step
fact_check_step = Step(
name="fact_check",
description="Verify facts and claims",
agent=fact_checker,
)
write_article = Step(
name="write_article",
description="Write final article",
agent=writer,
)
# === BASIC LINEAR WORKFLOW ===
basic_workflow = Workflow(
name="Basic Linear Workflow",
description="Research -> Summarize -> Condition(Fact Check) -> Write Article",
steps=[
research_step,
summarize_step,
Condition(
name="fact_check_condition",
description="Check if fact-checking is needed",
evaluator=needs_fact_checking,
steps=[fact_check_step],
),
write_article,
],
)
async def main():
try:
response: WorkflowRunOutput = await basic_workflow.arun(
input="量子コンピューティングにおける最近のブレークスルー",
)
pprint_run_response(response, markdown=True)
except Exception as e:
print(f"❌ Error: {e}")
if __name__ == "__main__":
asyncio.run(main())
出力例
INFO Executing async step (non-streaming): research INFO Setting default model to OpenAI Chat INFO Executing async step (non-streaming): summarize INFO Setting default model to OpenAI Chat INFO Executing async step (non-streaming): fact_check INFO Setting default model to OpenAI Chat INFO Executing async step (non-streaming): write_article INFO Setting default model to OpenAI Chat ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ 量子コンピューティングの最近の進展は、多くのニュース記事で報じられており、その革命的な可能性と、新たな安全保障上の課題への対応が求められていま… │ │ │ │ 1 量子コンピューティングの進展: │ │ • Googleの最新の「Willow」量子チップは、性能が従来のモデルの13,000倍速いとされ、量子コンピューティング分野における大きなマイルストーンを示… │ │ • 米国エネルギー省は、国の研究所での量子研究の推進のために6億2500万ドルを割り当て、量子情報科学の進展を目指しています。 │ │ • 超伝導量子ビット（qubits）は、実用的な量子プロセッサのプラットフォームとして有望視されており、量子コンピューティング技術の重要な進展を示… │ │ 2 ロンドンの技術イベントと政府の支援: │ │ • イギリス政府は、健康、国防、インフラなどの分野での量子技術の実世界での応用を加速させるために、1,400万ポンド以上を投資しています。これに … │ │ • The Economist Impactの「Commercialising Quantum │ │ Global」イベントでは、IonQが分野における発展を発表し、国際的な共同作業と商業的な関心を強調しました。 │ │ 3 量子コンピューティングとビットコインセキュリティ: │ │ • 量子コンピューティングは、ビットコインの暗号セキュリティに潜在的なリスクをもたらすとされ、2028年までにその脅威が現実化する可能性がありま… │ │ • SegWitウォレットに関する議論では、量子攻撃に対して限定的な保護を提供する可能性が示唆されており、システムの継続的なアップグレードとより強… │ │ │ │ これらの発展は、さまざまな産業を革命的に変える可能性がある一方で、新たな安全保障上の課題に対処するための戦略を必要としています。量子コンピュー… │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯
ストリーミング・レスポンス
ストリーミングを有効にするには、run() を呼び出すときに stream=True を設定します。これは単一のレスポンスではなく、WorkflowRunOutputEvent オブジェクトのイテレータを返します。
# Define your agents/team
...
content_creation_workflow = Workflow(
name="Content Creation Workflow",
description="Automated content creation from blog posts to social media",
db=SqliteDb(db_file="tmp/workflow.db"),
steps=[research_team, content_planner],
)
# Create and use workflow
if __name__ == "__main__":
response: Iterator[WorkflowRunOutputEvent] = content_creation_workflow.run(
input="AI trends in 2024",
markdown=True,
stream=True,
)
pprint_run_response(response, markdown=True)
すべてのイベントのストリーミング
レスポンスをストリーミングする際、デフォルトでは、WorkflowStartedEvent と WorkflowCompletedEvent イベントだけがストリーミングされます (すべてのエージェント及びチームのイベントも含まれます)。
stream_events=True を設定してすべてのイベントをストリーミングすることもできます。
これは、ワークフローの内部プロセスについてリアルタイムの更新情報を提供します :
from typing import Iterator
from agno.agent import Agent
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.workflow import Condition, Step, Workflow, StepInput
from agno.run.workflow import WorkflowRunOutput, WorkflowRunOutputEvent, WorkflowRunEvent
# === BASIC AGENTS ===
researcher = Agent(
name="Researcher",
instructions="Research the given topic and provide detailed findings.",
tools=[DuckDuckGoTools()],
)
summarizer = Agent(
name="Summarizer",
instructions="Create a clear summary of the research findings.",
)
fact_checker = Agent(
name="Fact Checker",
instructions="Verify facts and check for accuracy in the research.",
tools=[DuckDuckGoTools()],
)
writer = Agent(
name="Writer",
instructions="Write a comprehensive article based on all available research and verification.",
)
# === CONDITION EVALUATOR ===
def needs_fact_checking(step_input: StepInput) -> bool:
"""Determine if the research contains claims that need fact-checking"""
summary = step_input.previous_step_content or ""
# Look for keywords that suggest factual claims
fact_indicators = [
"study shows",
"breakthroughs",
"research indicates",
"according to",
"statistics",
"data shows",
"survey",
"report",
"million",
"billion",
"percent",
"%",
"increase",
"decrease",
]
return any(indicator in summary.lower() for indicator in fact_indicators)
# === WORKFLOW STEPS ===
research_step = Step(
name="research",
description="Research the topic",
agent=researcher,
)
summarize_step = Step(
name="summarize",
description="Summarize research findings",
agent=summarizer,
)
# Conditional fact-checking step
fact_check_step = Step(
name="fact_check",
description="Verify facts and claims",
agent=fact_checker,
)
write_article = Step(
name="write_article",
description="Write final article",
agent=writer,
)
# === BASIC LINEAR WORKFLOW ===
basic_workflow = Workflow(
name="Basic Linear Workflow",
description="Research -> Summarize -> Condition(Fact Check) -> Write Article",
steps=[
research_step,
summarize_step,
Condition(
name="fact_check_condition",
description="Check if fact-checking is needed",
evaluator=needs_fact_checking,
steps=[fact_check_step],
),
write_article,
],
)
if __name__ == "__main__":
try:
response: Iterator[WorkflowRunOutputEvent] = basic_workflow.run(
input="Recent breakthroughs in quantum computing",
stream=True,
stream_events=True,
)
for event in response:
if event.event == WorkflowRunEvent.condition_execution_started.value:
print(event)
print()
elif event.event == WorkflowRunEvent.condition_execution_completed.value:
print(event)
print()
elif event.event == WorkflowRunEvent.workflow_started.value:
print(event)
print()
elif event.event == WorkflowRunEvent.step_started.value:
print(event)
print()
elif event.event == WorkflowRunEvent.step_completed.value:
print(event)
print()
elif event.event == WorkflowRunEvent.workflow_completed.value:
print(event)
print()
except Exception as e:
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
Executor イベントのストリーミング
ワークフロー内で使用されるエージェントとチームのイベントはワークフローのストリーミングの最中に自動的に yield されます。stream_executor_events=False を設定してこれらの executor イベントをストリーミングしないように選択できます。
以下のワークフローイベントはすべての場合にストリーミングされます :
- WorkflowStarted
- WorkflowCompleted
- StepStarted
- StepCompleted
以下の例をご覧ください :
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow
agent = Agent(
name="ResearchAgent",
model=OpenAIChat(id="gpt-5-mini"),
instructions="You are a helpful research assistant. Be concise.",
)
workflow = Workflow(
name="Research Workflow",
steps=[Step(name="Research", agent=agent)],
stream=True,
stream_executor_events=False, # <- Filter out internal executor events
)
print("\n" + "=" * 70)
print("Workflow Streaming Example: stream_executor_events=False")
print("=" * 70)
print(
"\nThis will show only workflow and step events and will not yield RunContent and TeamRunContent events"
)
print("filtering out internal agent/team events for cleaner output.\n")
# Run workflow and display events
for event in workflow.run(
"What is Python?",
stream=True,
stream_events=True,
):
event_name = event.event if hasattr(event, "event") else type(event).__name__
print(f" → {event_name}")
非同期ストリーミング
Workflow.arun(stream=True) は単一のレスポンスではなく、WorkflowRunOutputEvent オブジェクトの非同期イテレータを返します。したがって例えば、レスポンスをストリーミングしたい場合、以下のようにできます :
# Define your workflow
...
async def main():
try:
response: AsyncIterator[WorkflowRunOutputEvent] = basic_workflow.arun(
message="Recent breakthroughs in quantum computing",
stream=True,
stream_events=True,
)
async for event in response:
if event.event == WorkflowRunEvent.condition_execution_started.value:
print(event)
print()
elif event.event == WorkflowRunEvent.condition_execution_completed.value:
print(event)
print()
elif event.event == WorkflowRunEvent.workflow_started.value:
print(event)
print()
elif event.event == WorkflowRunEvent.step_started.value:
print(event)
print()
elif event.event == WorkflowRunEvent.step_completed.value:
print(event)
print()
elif event.event == WorkflowRunEvent.workflow_completed.value:
print(event)
print()
except Exception as e:
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(main())
以上