Skip to content

NatsAdapter

fastagency.adapters.nats.base.NatsAdapter #

NatsAdapter(
    provider: ProviderProtocol,
    *,
    nats_url: Optional[str] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
    super_conversation: Optional[NatsAdapter] = None
)

Bases: MessageProcessorMixin, CreateWorkflowUIMixin

Provider for NATS.

PARAMETER DESCRIPTION
provider

The provider.

TYPE: ProviderProtocol

nats_url

The NATS URL. Defaults to None in which case 'nats://localhost:4222' is used.

TYPE: Optional[str] DEFAULT: None

user

The user. Defaults to None.

TYPE: Optional[str] DEFAULT: None

password

The password. Defaults to None.

TYPE: Optional[str] DEFAULT: None

super_conversation

The super conversation. Defaults to None.

TYPE: Optional[NatsProvider] DEFAULT: None

Source code in fastagency/adapters/nats/base.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def __init__(
    self,
    provider: ProviderProtocol,
    *,
    nats_url: Optional[str] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
    super_conversation: Optional["NatsAdapter"] = None,
) -> None:
    """Provider for NATS.

    Args:
        provider (ProviderProtocol): The provider.
        nats_url (Optional[str], optional): The NATS URL. Defaults to None in which case 'nats://localhost:4222' is used.
        user (Optional[str], optional): The user. Defaults to None.
        password (Optional[str], optional): The password. Defaults to None.
        super_conversation (Optional["NatsProvider"], optional): The super conversation. Defaults to None.
    """
    self.provider = provider
    self.nats_url = nats_url or "nats://localhost:4222"
    self.user = user
    self.password = password
    self.queue: Queue = Queue()  # type: ignore[type-arg]

    self.broker = NatsBroker(self.nats_url, user=user, password=password)
    self.app = FastStream(self.broker)
    self.subscriber: "AsyncAPISubscriber"
    self._input_request_subject: str
    self._input_receive_subject: str

    self.super_conversation: Optional[NatsAdapter] = super_conversation
    self.sub_conversations: list[NatsAdapter] = []

    self._create_initiate_subscriber()

app instance-attribute #

app = FastStream(broker)

broker instance-attribute #

broker = NatsBroker(nats_url, user=user, password=password)

nats_url instance-attribute #

nats_url = nats_url or 'nats://localhost:4222'

password instance-attribute #

password = password

provider instance-attribute #

provider = provider

queue instance-attribute #

queue: Queue = Queue()

sub_conversations instance-attribute #

sub_conversations: list[NatsAdapter] = []

subscriber instance-attribute #

subscriber: AsyncAPISubscriber

super_conversation instance-attribute #

super_conversation: Optional[NatsAdapter] = (
    super_conversation
)

user instance-attribute #

user = user

create #

create(app: Runnable, import_string: str) -> Iterator[None]
Source code in fastagency/adapters/nats/base.py 
360
361
362
@contextmanager
def create(self, app: Runnable, import_string: str) -> Iterator[None]:
    raise NotImplementedError("NatsAdapter.create() is not implemented")

create_provider classmethod #

create_provider(
    nats_url: Optional[str] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
) -> ProviderProtocol
Source code in fastagency/adapters/nats/base.py 
351
352
353
354
355
356
357
358
@classmethod
def create_provider(
    cls,
    nats_url: Optional[str] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
) -> ProviderProtocol:
    return NatsProvider(nats_url=nats_url, user=user, password=password)

create_subconversation #

create_subconversation() -> NatsAdapter
Source code in fastagency/adapters/nats/base.py 
348
349
def create_subconversation(self) -> "NatsAdapter":
    return self

create_workflow_ui #

create_workflow_ui(workflow_uuid: str) -> UI
Source code in fastagency/base.py 
59
60
def create_workflow_ui(self: UIBase, workflow_uuid: str) -> "UI":
    return UI(uibase=self, workflow_uuid=workflow_uuid)

error #

error(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    short: Optional[str] = None,
    long: Optional[str] = None,
) -> Optional[str]
Source code in fastagency/messages.py 
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
def error(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    short: Optional[str] = None,
    long: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        Error(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            short=short,
            long=long,
        )
    )

function_call_execution #

function_call_execution(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    function_name: Optional[str] = None,
    call_id: Optional[str] = None,
    retval: Any = None,
) -> Optional[str]
Source code in fastagency/messages.py 
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
def function_call_execution(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    function_name: Optional[str] = None,
    call_id: Optional[str] = None,
    retval: Any = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        FunctionCallExecution(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            function_name=function_name,
            call_id=call_id,
            retval=retval,
        )
    )

keep_alive #

keep_alive(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
) -> Optional[str]
Source code in fastagency/messages.py 
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
def keep_alive(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        KeepAlive(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
        )
    )

lifespan async #

lifespan(app: Any) -> AsyncIterator[None]
Source code in fastagency/adapters/nats/base.py 
238
239
240
241
242
243
244
245
246
@asynccontextmanager
async def lifespan(self, app: Any) -> AsyncIterator[None]:
    async with self.broker:
        await self.broker.start()
        await self._publish_discovery()
        try:
            yield
        finally:
            await self.broker.close()

multiple_choice #

multiple_choice(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    prompt: Optional[str] = None,
    choices: Optional[list[str]] = None,
    default: Optional[str] = None,
    single: bool = True,
) -> Optional[str]
Source code in fastagency/messages.py 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
def multiple_choice(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    prompt: Optional[str] = None,
    choices: Optional[list[str]] = None,
    default: Optional[str] = None,
    single: bool = True,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    choices = choices or []
    return self.process_message(
        MultipleChoice(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            prompt=prompt,
            choices=choices,
            default=default,
            single=single,
        )
    )

process_message #

process_message(message: IOMessage) -> Optional[str]
Source code in fastagency/adapters/nats/base.py 
338
339
340
341
342
343
344
345
346
def process_message(self, message: IOMessage) -> Optional[str]:
    try:
        return self.visit(message)
    except Exception as e:
        logger.error(f"Error in process_message: {e}", stack_info=True)
        # do not reraise, we must go on
        if isinstance(message, AskingMessage):
            return "Error: Something went wrong. Please check logs for details."
        return None

start #

start(
    *,
    app: Runnable,
    import_string: str,
    name: Optional[str] = None,
    params: dict[str, Any],
    single_run: bool = False
) -> None
Source code in fastagency/adapters/nats/base.py 
364
365
366
367
368
369
370
371
372
373
def start(
    self,
    *,
    app: Runnable,
    import_string: str,
    name: Optional[str] = None,
    params: dict[str, Any],
    single_run: bool = False,
) -> None:
    raise NotImplementedError("NatsAdapter.start() is not implemented")

suggested_function_call #

suggested_function_call(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    function_name: Optional[str] = None,
    call_id: Optional[str] = None,
    arguments: Optional[dict[str, Any]] = None,
) -> Optional[str]
Source code in fastagency/messages.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
def suggested_function_call(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    function_name: Optional[str] = None,
    call_id: Optional[str] = None,
    arguments: Optional[dict[str, Any]] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    arguments = arguments or {}
    return self.process_message(
        SuggestedFunctionCall(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            function_name=function_name,
            call_id=call_id,
            arguments=arguments,
        )
    )

system_message #

system_message(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    message: Optional[dict[str, Any]] = None,
) -> Optional[str]
Source code in fastagency/messages.py 
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
def system_message(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    message: Optional[dict[str, Any]] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    message = message or {}
    return self.process_message(
        SystemMessage(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            message=message,
        )
    )

text_input #

text_input(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    prompt: Optional[str] = None,
    suggestions: Optional[list[str]] = None,
    password: bool = False,
) -> Optional[str]
Source code in fastagency/messages.py 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
def text_input(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    prompt: Optional[str] = None,
    suggestions: Optional[list[str]] = None,
    password: bool = False,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    suggestions = suggestions or []
    return self.process_message(
        TextInput(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            prompt=prompt,
            suggestions=suggestions,
            password=password,
        )
    )

text_message #

text_message(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    body: Optional[str] = None,
) -> Optional[str]
Source code in fastagency/messages.py 
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
def text_message(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    body: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        TextMessage(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            body=body,
        )
    )

visit #

visit(message: IOMessage) -> Optional[str]
Source code in fastagency/messages.py 
318
319
320
321
def visit(self, message: IOMessage) -> Optional[str]:
    method_name = f"visit_{message.type}"
    method = getattr(self, method_name, self.visit_default)
    return method(message)

visit_default #

visit_default(message: IOMessage) -> None
Source code in fastagency/adapters/nats/base.py 
248
249
250
251
def visit_default(self, message: IOMessage) -> None:
    content = message.model_dump()
    logger.debug(f"visit_default(): {content=}")
    syncify(self.broker.publish)(content, self._input_request_subject)

visit_error #

visit_error(message: Error) -> Optional[str]
Source code in fastagency/messages.py 
357
358
def visit_error(self, message: Error) -> Optional[str]:
    return self.visit_default(message)

visit_function_call_execution #

visit_function_call_execution(
    message: FunctionCallExecution,
) -> Optional[str]
Source code in fastagency/messages.py 
334
335
336
337
def visit_function_call_execution(
    self, message: FunctionCallExecution
) -> Optional[str]:
    return self.visit_default(message)

visit_keep_alive #

visit_keep_alive(message: KeepAlive) -> Optional[str]
Source code in fastagency/messages.py 
348
349
def visit_keep_alive(self, message: KeepAlive) -> Optional[str]:
    return self.visit_default(message)

visit_multiple_choice #

visit_multiple_choice(message: MultipleChoice) -> str
Source code in fastagency/adapters/nats/base.py 
326
327
328
329
330
331
332
333
334
335
336
def visit_multiple_choice(self, message: MultipleChoice) -> str:
    content = message.model_dump()
    question_id = message.uuid
    logger.info(f"visit_multiple_choice(): {content=}")
    syncify(self.broker.publish)(content, self._input_request_subject)

    input_response: InputResponseModel = syncify(
        self._wait_for_question_response_with_timeout
    )(question_id=question_id)
    logger.info(input_response)
    return input_response.msg

visit_suggested_function_call #

visit_suggested_function_call(
    message: SuggestedFunctionCall,
) -> Optional[str]
Source code in fastagency/messages.py 
329
330
331
332
def visit_suggested_function_call(
    self, message: SuggestedFunctionCall
) -> Optional[str]:
    return self.visit_default(message)

visit_system_message #

visit_system_message(
    message: SystemMessage,
) -> Optional[str]
Source code in fastagency/messages.py 
345
346
def visit_system_message(self, message: SystemMessage) -> Optional[str]:
    return self.visit_default(message)

visit_text_input #

visit_text_input(message: TextInput) -> str
Source code in fastagency/adapters/nats/base.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def visit_text_input(self, message: TextInput) -> str:
    content = message.model_dump()
    question_id = message.uuid
    logger.info(f"visit_text_input(): {content=}")
    syncify(self.broker.publish)(content, self._input_request_subject)
    logger.info(
        f"visit_text_input(): published message '{content}' to {self._input_request_subject}"
    )

    input_response: InputResponseModel = syncify(
        self._wait_for_question_response_with_timeout
    )(question_id=question_id)
    logger.info(input_response)
    return input_response.msg

visit_text_message #

visit_text_message(message: TextMessage) -> None
Source code in fastagency/adapters/nats/base.py 
253
254
255
256
def visit_text_message(self, message: TextMessage) -> None:
    content = message.model_dump()
    logger.debug(f"visit_text_message(): {content=}")
    syncify(self.broker.publish)(content, self._input_request_subject)

visit_workflow_completed #

visit_workflow_completed(
    message: WorkflowCompleted,
) -> Optional[str]
Source code in fastagency/messages.py 
354
355
def visit_workflow_completed(self, message: WorkflowCompleted) -> Optional[str]:
    return self.visit_default(message)

visit_workflow_started #

visit_workflow_started(
    message: WorkflowStarted,
) -> Optional[str]
Source code in fastagency/messages.py 
351
352
def visit_workflow_started(self, message: WorkflowStarted) -> Optional[str]:
    return self.visit_default(message)

workflow_completed #

workflow_completed(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    result: Optional[str] = None,
) -> Optional[str]
Source code in fastagency/messages.py 
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
def workflow_completed(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    result: Optional[str] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    return self.process_message(
        WorkflowCompleted(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            result=result,
        )
    )

workflow_started #

workflow_started(
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    params: Optional[dict[str, Any]] = None,
) -> Optional[str]
Source code in fastagency/messages.py 
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
def workflow_started(
    self,
    workflow_uuid: str,
    sender: Optional[str] = None,
    recipient: Optional[str] = None,
    auto_reply: bool = False,
    uuid: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    params: Optional[dict[str, Any]] = None,
) -> Optional[str]:
    uuid = uuid or str(uuid4().hex)
    params = params or {}
    return self.process_message(
        WorkflowStarted(
            sender=sender,
            recipient=recipient,
            auto_reply=auto_reply,
            uuid=uuid,
            workflow_uuid=workflow_uuid,
            name=name,
            description=description,
            params=params,
        )
    )