98 lines
2.5 KiB
Python
98 lines
2.5 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.connection = connection
|
|
|
|
# Submit a request for info, return response.
|
|
# Must disclose what service it's for.
|
|
async def fetch(self, entities, service):
|
|
|
|
# Assert list.
|
|
if type(entities) == str:
|
|
entities = [entities]
|
|
|
|
print(f"Fetching {[e for e in entities]} for {service}")
|
|
|
|
# Build intent.
|
|
params = {
|
|
"service": service,
|
|
"entities": entities
|
|
}; intent = typs.Intent("fetch", params)
|
|
|
|
# Send intent to agent, wait for response.
|
|
await self.send_agent_intent(intent)
|
|
response = await self.recv_agent_intent()
|
|
|
|
# Parse response and return entities.
|
|
entities = response.params.get("found")
|
|
print(f"Fetch response: {entities}")
|
|
return entities
|
|
|
|
# Text the client something. Also, let agent know
|
|
# when your service is complete via the complete
|
|
# attribute.
|
|
# Must disclose what service it's for.
|
|
async def notify(self, text, end=False):
|
|
|
|
print(f"Notifying: {text} for {self.service}")
|
|
|
|
params = {
|
|
"service": self.service,
|
|
"text": text,
|
|
"end": end
|
|
}
|
|
|
|
# Build the intent
|
|
intent = typs.Intent("notify", params)
|
|
|
|
# Send intent to agent.
|
|
await self.send_agent_intent(intent)
|
|
|
|
# Get user info.
|
|
async def sync(self):
|
|
|
|
intent = typs.Intent("sync", params={})
|
|
|
|
# Send intent to agent.
|
|
await self.send_agent_intent(intent)
|
|
|
|
# Recv and return profile.
|
|
intent = await self.recv_agent_intent()
|
|
profile = intent.params.get("profile")
|
|
return profile
|
|
|
|
async def recv_agent_intent(self):
|
|
package = await self.connection.recv()
|
|
intent = self.__decode(package)
|
|
return intent
|
|
|
|
async def send_agent_intent(self, intent):
|
|
package = self.__encode(intent)
|
|
await self.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
|
|
|
|
|
|
|
|
|