i'0SSKJr SSKJrJrJr SSKJrJrJ r SSK J r J r J r JrJr SSKJr "SS\ 5r"SS \5r"S S \5r"S S \5r"SS\5r"SS\5r"SS\5r"SS\5r"SS\5r\\rg)) annotations) dict_items dict_keys dict_values)AnyTypeOptional) BaseModel ConfigDictField PrivateAttrmodel_serializer)Enumc^\rSrSr%Sr\"SS9r\"\S9r S\ S'SU4Sjjr SU4S jjr SU4S jjr SS jrSS jrSS S jjrS!SjrS"SjrS#SjrS$SjrS%SjrS&SjrS'SjrS(Sjr\"SS9S)Sj5rSrU=r$)* DictLikeModela Base Pydantic model class that mimics a dict interface for dynamic fields. Known, typed fields behave like regular Pydantic attributes. Any extra keyword arguments are stored in an internal dict and can be accessed through both attribute and mapping semantics. This hybrid model enables flexible event payloads while preserving validation for declared fields. PrivateAttr: _data (dict[str, Any]): Underlying Python dict for dynamic fields. T)arbitrary_types_allowed)default_factorydict[str, Any]_datac j>0n0n0nUR5H=upVXPRR;aXbU'M$XPR;aXcU'M9XdU'M? [T U]"S0UD6 UR5Hupx[T U]Xx5 M U(aURRU5 gg)zK __init__. NOTE: fields and private_attrs are pulled from params by name. N) items __class__ model_fields__private_attributes__super__init__ __setattr__rupdate) selfparamsfields private_attrsdatakv private_attrvaluers J/home/james-whalen/.local/lib/python3.13/site-packages/workflows/events.pyrDictLikeModel.__init__#s LLNDANN///q 111#$a Q # "6"#0#6#6#8 L G  4$9  JJ  d # c>XR;dXRR;a[TU]U5$XR ;a&[ SURRSUS35eUR U$)N'z' object has no attribute ')rrrr __getattr__rAttributeError__name__)r!_DictLikeModel__namers r*r/DictLikeModel.__getattr__:sv 11 14447&v. .ZZ'$//00KF8STU::f% %r,c>XR;dXRR;a[TU]X5 gUR R X5 gN)rrrrrr __setitem__)r!namer)rs r*rDictLikeModel.__setattr__Gs= .. .$..:U:U2U G  , JJ " "4 /r,c URU$r5rr!keys r* __getitem__DictLikeModel.__getitem__Mszz#r,c X RU'gr5r:)r!r<r)s r*r6DictLikeModel.__setitem__Ps  3r,c8URRX5$r5)rget)r!r<defaults r*rBDictLikeModel.getSszz~~c++r,cXR;$r5r:r;s r* __contains__DictLikeModel.__contains__Vsjj  r,c6URR5$r5)rkeysr!s r*rIDictLikeModel.keysYszz  r,c6URR5$r5)rvaluesrJs r*rMDictLikeModel.values\szz  ""r,c6URR5$r5)rrrJs r*rDictLikeModel.items_szz!!r,c,[UR5$r5)lenrrJs r*__len__DictLikeModel.__len__bs4::r,c,[UR5$r5)iterrrJs r*__iter__DictLikeModel.__iter__esDJJr,cUR$r5r:)r!argskwargss r*to_dictDictLikeModel.to_dicths zzr,cg)z.Make test `if event:` pass on Event instances.TrrJs r*__bool__DictLikeModel.__bool__ksr,wrapmodecVU"U5nUR(aURUS'U$)Nrr:r!handlerr%s r*custom_model_dumpDictLikeModel.custom_model_dumpos$t} :: JJDM r,rr"r)r2strreturnr)r7rjr)rrkNone)r<rjrkr)r<rjr)rrkrlr5)r<rjrCrrkr)r<rjrkbool)rkz'dict_keys[str, Any]')rkz'dict_values[str, Any]')rkz'dict_items[str, Any]')rkintrkr)rZrr[rrkr)rkrmrfrrkr)r1 __module__ __qualname____firstlineno____doc__r model_configr dictr__annotations__rr/rr=r6rBrFrIrMrrSrWr\r_rrg__static_attributes__ __classcell__rs@r*rrs d;L'=E>=$. &0  ,!!#" 6"#r,rc0^\rSrSrSrSU4SjjrSrU=r$)Eventxaf Base class for all workflow events. Events are light-weight, serializable payloads passed between steps. They support both attribute and mapping access to dynamic fields. Examples: Subclassing with typed fields: ```python from pydantic import Field class CustomEv(Event): score: int = Field(ge=0) e = CustomEv(score=10) print(e.score) ``` See Also: - [StartEvent][workflows.events.StartEvent] - [StopEvent][workflows.events.StopEvent] - [InputRequiredEvent][workflows.events.InputRequiredEvent] - [HumanResponseEvent][workflows.events.HumanResponseEvent] c &>[TU]"S0UD6 g)Nrrr)r!r"rs r*rEvent.__init__s "6"r,rri)r1rqrrrsrtrrxryrzs@r*r|r|xs4##r,r|c\rSrSrSrSrg) StartEventz9Implicit entry event sent to kick off a `Workflow.run()`.rNr1rqrrrsrtrxrr,r*rrsCr,rc^\rSrSr%Sr\"SS9rS\S'SSU4SjjjrSSjr \ SS j5r \ "S S 9SS j5r SS jrSSjrSrU=r$) StopEventaCTerminal event that signals the workflow has completed. The `result` property contains the return value of the workflow run. When a custom stop event subclass is used, the workflow result is that event instance itself. Examples: ```python # default stop event: result holds the value return StopEvent(result={"answer": 42}) ``` Subclassing to provide a custom result: ```python class MyStopEv(StopEvent): pass @step async def my_step(self, ctx: Context, ev: StartEvent) -> MyStopEv: return MyStopEv(result={"answer": 42}) N)rCr_resultc *>[TU]"SSU0UD6 g)Nrrr)r!resultr[rs r*rStopEvent.__init__s 2262r,cUR$)zBThis can be overridden by subclasses to return the desired result.rrJs r* _get_resultStopEvent._get_results ||r,c"UR5$r5)rrJs r*rStopEvent.results!!r,rarbcNU"U5nURbURUS'U$)Nrrres r*rgStopEvent.custom_model_dumps't} << #!\\DN r,c0UREUR5EnUR"5VVs/sH up#USU<3PM nnnSRU5nURR SUS3$s snnf)N=z, ())r model_dumprjoinrr1)r!rr&r'partsdict_strs r*__repr__StopEvent.__repr__sx8 8doo&78 *4*:*:*<=*<$!A3au*<=99U#..))*!H:Q77>sA7c,[UR5$r5)rjrrJs r*__str__StopEvent.__str__s4<<  r,rr5)rrr[rrkrlrorp)rkrj)r1rqrrrsrtr rrwrrpropertyrrrgrrrxryrzs@r*rrsd.t,GS,33""6"#8!!r,rc\rSrSrSrSrg)InputRequiredEventaEmitted when human input is required to proceed. Automatically written to the event stream if returned from a step. If returned from a step, it does not need to be consumed by other steps and will pass validation. It's expected that the caller will respond to this event and send back a [HumanResponseEvent][workflows.events.HumanResponseEvent]. Use this directly or subclass it. Typical flow: a step returns `InputRequiredEvent`, callers consume it from the stream and send back a [HumanResponseEvent][workflows.events.HumanResponseEvent]. Examples: ```python from workflows.events import InputRequiredEvent, HumanResponseEvent class HITLWorkflow(Workflow): @step async def my_step(self, ev: StartEvent) -> InputRequiredEvent: return InputRequiredEvent(prefix="What's your name? ") @step async def my_step(self, ev: HumanResponseEvent) -> StopEvent: return StopEvent(result=ev.response) ``` rNrrr,r*rrsr,rc\rSrSrSrSrg)HumanResponseEventa3Carries a human's response for a prior input request. If consumed by a step and not returned by another, it will still pass validation. Examples: ```python from workflows.events import InputRequiredEvent, HumanResponseEvent class HITLWorkflow(Workflow): @step async def my_step(self, ev: StartEvent) -> InputRequiredEvent: return InputRequiredEvent(prefix="What's your name? ") @step async def my_step(self, ev: HumanResponseEvent) -> StopEvent: return StopEvent(result=ev.response) ``` rNrrr,r*rrsr,rc\rSrSrSrSrg)InternalDispatchEventia InternalDispatchEvent is a special event type that exposes processes running inside workflow, even if the user did not explicitly expose them by setting, e.g., `ctx.write_event_to_stream(`. Examples: ```python wf = ExampleWorkflow() handler = wf.run(message="Hello, who are you?") async for ev in handler.stream_event(expose_internal=True): if isinstance(ev, InternalDispatchEvent): print(type(ev), ev) ``` rNrrr,r*rrs   r,rc \rSrSrSrSrSrSrg) StepStatei preparingrunning not_runningrN)r1rqrrrs PREPARINGRUNNING NOT_RUNNINGrxrr,r*rrsIGKr,rc\rSrSr%Sr\"SS9rS\S'\"SS9rS\S '\"S S9r S\S '\"S S9r S\S '\"SSS9r S\S'Sr g)StepStateChangedi!a StepStateChanged is a special event type that exposes internal changes in the state of the event, including whether the step is running or in progress, what worker it is running on and what events it takes as input and output, as well as changes in the workflow state. Attributes: name (str): Name of the step step_state (StepState): State of the step ("running", "not_running", "in_progress", "not_in_progress", "exited") worker_id (str): ID of the worker that the step is running on input_event_name (str): Name of the input event output_event_name (Optional[str]): Name of the output event context_state (dict[str, Any]): Snapshot of the current workflow state zName of the step) descriptionrjr7zXState of the step ('running', 'not_running', 'in_progress', 'not_in_progress', 'exited')r step_statez,ID of the worker that the step is running on worker_idzName of the input eventinput_event_namezName of the output eventN)rrCz Optional[str]output_event_namer) r1rqrrrsrtr r7rwrrrrrxrr,r*rr!sg "45D#5!nJ 'UVIsV!.GHcH',.(}r,rN) __future__r_collections_abcrrrtypingrrr pydanticr r r r renumrrr|rrrrrrr EventTyperr,r*rs#??&&bIbJ#M#>DD6!6!r:* E $  ,0 K r,