AGUIStream - AG2

AGUIStream

``autogen.ag_ui.AGUIStream #

AGUIStream(agent, *, event_interceptors=())

Source code in autogen/ag_ui/adapter.py
<br>51<br>52<br>53<br>54<br>55<br>56<br>57<br>58<br>59<br> <br>def __init__(<br> self,<br> agent: ConversableAgent,<br> *,<br> event_interceptors: Iterable[Callable[[ServiceResponse], AsyncIterator[BaseEvent]]] = (),<br>) -> None:<br> self.__agent = agent<br> self.__event_interceptors = event_interceptors<br> self.service = AgentService(agent)<br>

``serviceinstance-attribute#

service = AgentService(agent)

``dispatchasync#

dispatch(incoming, *, context=None, accept=None)

Source code in autogen/ag_ui/adapter.py
<br>61<br>62<br>63<br>64<br>65<br>66<br>67<br>68<br>69<br>70<br>71<br>72<br>73<br>74<br>75<br>76<br>77<br>78<br>79<br>80<br>81<br>82<br>83<br>84<br>85<br>86<br>87<br>88<br>89<br>90<br>91<br>92<br>93<br> ```
async def dispatch(
self,
incoming: RunAgentInput,
*,
context: dict[str, Any]

``build_asgi #

build_asgi()

Build an ASGI endpoint for the AGUIStream.

Source code in autogen/ag_ui/adapter.py
<br> 95<br> 96<br> 97<br> 98<br> 99<br>100<br> <br>def build_asgi(self) -> "type[HTTPEndpoint]":<br> """Build an ASGI endpoint for the AGUIStream."""<br> # import here to avoid Starlette requirements in the main package<br> from .asgi import build_asgi<br> return build_asgi(self)<br>

Back to top