munggoggo.asgi_agent

Wraps agents into ASGI web application in order to expose their functionality via HTTP(S) or websocket. Support OPENAPI documentation of endpoints and provides jsonrpc facility.

https://github.com/encode/uvicorn/issues/183#issuecomment-515735604

In order to run message-consumers based on aio-pika inside a set of HTTP endpoints served by uvicorn in the same event loop, it requires to move configuration and initialization of consumers inside the app served by uvicorn.

class asgi_agent.AsgiAgent(agent, *args, **kwargs)

Wraps agent into Starlette ASGI application. Implements jsonrpc interface:

curl -X POST -H "content-type: application/json" -d '{"method":"example_rpc_method","params":[1,2],"jsonrpc":"2.0","id":1}' http://localhost:8000/jsonrpc

Provides OPENAPI (Swagger) documention of exposed endpoints via https://apispec.readthedocs.io/en/latest/:

http://0.0.0.0:8000/openapi

Serves static files from ./static:

http://0.0.0.0:8000/static/schema.yaml

Exposes websocket at:

http://0.0.0.0:8000/ws.

add_schema(name: str, schema: marshmallow.schema.Schema, check_existing: bool = True) → None

Adds a mashmallow schema to the API specification.

Parameters

• name – ClassName

• schema – SchemaName of Class

• check_existing – make sure it is only added once

async homepage(request)

home — description: Home responses:

200:

content:

application/json:

schema: PlatformInformationSchema

schema(name, **options)

Decorator for creating new routes around function and class definitions.

Caveat: agent instance must exist in order to use it

Usage:

@asgi_agent.schema("Pet")
class PetSchema(Schema):
    name = fields.Str()

class asgi_agent.JsonRpcSchema(*, only: Union[Sequence[str], Set[str]] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Dict = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: str = None)

class asgi_agent.PlatformInformationSchema(*, only: Union[Sequence[str], Set[str]] = None, exclude: Union[Sequence[str], Set[str]] = (), many: bool = False, context: Dict = None, load_only: Union[Sequence[str], Set[str]] = (), dump_only: Union[Sequence[str], Set[str]] = (), partial: Union[bool, Sequence[str], Set[str]] = False, unknown: str = None)

class asgi_agent.WsController(scope: MutableMapping[str, Any])

Implements websocket communication for AsgiAgent. Exposes websocket at Core as self.ws

async on_connect(websocket)

Override to handle an incoming websocket connection

async on_disconnect(websocket: starlette.websockets.WebSocket, close_code: int) → None

Override to handle a disconnecting websocket

async on_receive(websocket, data)

Override to handle an incoming websocket message