68 lines
1.7 KiB
Python
68 lines
1.7 KiB
Python
import json
|
|
import asyncio
|
|
import websockets
|
|
|
|
import typs
|
|
|
|
|
|
class Agent:
|
|
"""Session interface for Crimata Agent
|
|
|
|
Receive core functionalities for communicating with Crimata
|
|
Agent in an OOP way.
|
|
|
|
"""
|
|
def __init__(self, connection):
|
|
self.service = None
|
|
self.client_connection = connection
|
|
|
|
def fetch(self, entity, invocation):
|
|
# Create the intent.
|
|
params = {
|
|
"entity": entity,
|
|
"service": self.service,
|
|
"invocation": invocation
|
|
}
|
|
intent = typs.Intent(name="fetch", params=params)
|
|
|
|
self.send_agent_intent(intent)
|
|
intent = self.recv_agent_intent()
|
|
|
|
response = intent.params.get("response")
|
|
return response
|
|
|
|
def notify(self, text, end=False):
|
|
# Create the intent.
|
|
params = {
|
|
"text": text,
|
|
"service": "message",
|
|
"end": end
|
|
}
|
|
intent = typs.Intent(name="notify", params=params)
|
|
|
|
# Send intent to agent.
|
|
self.send_agent_intent(intent)
|
|
|
|
async def recv_agent_intent(self):
|
|
package = await self.client_connection.recv()
|
|
intent = self.__decode(package)
|
|
return intent
|
|
|
|
async def send_agent_intent(self, intent):
|
|
package = self.__encode(intent)
|
|
await self.client_connection.send(package)
|
|
|
|
def __encode(self, intent: typs.Intent):
|
|
package = json.dumps(intent.__dict__)
|
|
return package
|
|
|
|
def __decode(self, package) -> typs.Intent:
|
|
jpackage = json.loads(package)
|
|
name = jpackage.get("name")
|
|
params = jpackage.get("params")
|
|
intent = typs.Intent(name, params)
|
|
return intent
|
|
|
|
|
|
|
|
|