UMS-Agenten/Agenten-Plattform
3
0
Fork 0
Code Issues Actions Packages Releases Activity

Access Messages via API from Python

Browse Source
This commit is contained in:
Magnus Bender
2024-10-30 18:02:05 +01:00
parent 01db00b3b4
commit cfe3dbd5bb
13 changed files with 1665 additions and 823 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
ums/example/__main__.py
View File

@ -8,35 +8,32 @@
# source code released under the terms of GNU Public License Version 3
# https://www.gnu.org/licenses/gpl-3.0.txt
"""
See the source →
"""
if __name__ == "__main__":
## Example: Sending messages to management via python
from ums.utils import ManagementRequest
from ums.utils import AgentMessage, RiddleData, RiddleDataType, RiddleSolution, ManagementRequest
m_request = ManagementRequest()
ex = AgentMessage(
id="ex5",
riddle={
"context":"Example 1",
"question":"Get the name of the person."
},
data=[
RiddleData(
type=RiddleDataType.TEXT,
file_plain="./cv.txt"
)
]
)
ex.status.extract.required = False
# get info from Management
ex.solution = RiddleSolution(
solution="Otto",
explanation="Written in line 6 after 'Name:'"
print(
m_request.get_message(count=12)
)
mr = ManagementRequest("localhost")
print(
m_request.list_messages(id="test", limit=2)
)
print(mr.send_message(ex))
print(mr.get_status(20))
print(
m_request.total_messages(id="test")
)
from ums.utils import AgentMessage, RiddleData, RiddleDataType, RiddleSolution
# send messages to management
# TODO

2
ums/management/interface.py
View File

@ -31,7 +31,7 @@ class Interface():
self.router = APIRouter(
prefix=self._PREFIX,
tags=["app, gui"]
tags=["gui"]
)
self._add_routes()

29
ums/management/main.py
View File

@ -10,6 +10,7 @@
import os
from typing import List
from datetime import datetime
from fastapi import FastAPI, Request, BackgroundTasks, HTTPException
@ -78,7 +79,7 @@ class WebMain():
{"request" : request}
)
@self.app.post("/message", summary="Send a message to the management")
@self.app.post("/message", summary="Send a message to the management", tags=['agents'])
def message(request: Request, message:AgentMessage, background_tasks: BackgroundTasks) -> AgentResponse:
receiver = request.headers['host']
@ -89,14 +90,36 @@ class WebMain():
return self.msg_process.new_message(sender, receiver, message, background_tasks)
@self.app.get("/status", summary="Get status of a message")
@self.app.get("/list", summary="Get list of messages (like table)", tags=["cli, agents"])
def list(id:str|None=None, sender:str|None=None, recipient:str|None=None,
processed:bool|None=None, solution:bool|None=None,
time_after:int|None=None, time_before:int|None=None,
limit:int=10, offset:int=0
) -> List[MessageDbRow]:
db_args = {
"limit" : limit,
"offset" : offset
}
for v,n in (
(id,'id'), (sender,'sender'), (recipient,'recipient'),
(processed,'processed'), (solution,'solution'),
(time_after, 'time_after'), (time_before, 'time_before')
):
if not v is None:
db_args[n] = v
return [row for row in self.db.iterate(**db_args)]
@self.app.get("/list/single", summary="Get a single message", tags=["cli, agents"])
def status(count:int) -> MessageDbRow:
msg = self.db.by_count(count)
if msg is None:
raise HTTPException(status_code=404, detail="Message not found")
return msg
if __name__ == "ums.management.main" and os.environ.get('SERVE', 'false') == 'true':
main = WebMain()
app = main.app

129
ums/utils/request.py
View File

@ -8,31 +8,140 @@
# source code released under the terms of GNU Public License Version 3
# https://www.gnu.org/licenses/gpl-3.0.txt
"""
Access to the management, e.g., get the list of messages and single messages.
Manually send messages (if necessary, the platforms should do this).
### Example
```python
m_request = ManagementRequest()
m_request.get_message(count=12)
# MessageDbRow(count=12 sender='from' recipient='to' ...
m_request.list_messages(id="test", limit=2)
# [
# MessageDbRow(count=7256, sender='management', ...),
# MessageDbRow(count=7255, sender='management', ...),
# ]
m_request.total_messages(id="test")
# 31
```
See also `ums.example.__main__` and run in Docker via ``docker compose exec management python -m ums.example``
"""
import os
from typing import List, Dict, Any
import requests
from pydantic import validate_call
from ums.utils.types import AgentMessage, AgentResponse, MessageDbRow
class RequestException(Exception):
pass
"""
Raised on http and similar errors.
"""
pass
class ManagementRequest():
def __init__(self, hostname:str, port:int=80):
self.url = "http://{hostname}:{port}".format(hostname=hostname, port=port)
MANAGEMENT_URL = os.environ.get('MANAGEMENT_URL', 'http://127.0.0.1:80').strip().strip('/')
def get_status(self, count:int) -> MessageDbRow:
@validate_call
def __init__(self, allow_lazy:bool=True):
"""
If `allow_lazy` is active, the type checking (by pydantic) is less strict.
E.g. it does not require that all files in the data section of messages must exist on the file system.
"""
self.allow_lazy = allow_lazy
self.pydantic_context = {
"require_file_exists": not self.allow_lazy
}
@validate_call
def get_message(self, count:int) -> MessageDbRow:
"""
Get a message (like a table row) from the management by using the `count`.
"""
row = self._get_request(
'list/single',
{"count": count}
)
return MessageDbRow.model_validate(
row, context=self.pydantic_context
)
@validate_call
def list_messages(self,
id:str|None=None, sender:str|None=None, recipient:str|None=None,
processed:bool|None=None, solution:bool|None=None,
time_after:int|None=None, time_before:int|None=None,
limit:int=10, offset:int=0
) -> List[MessageDbRow]:
"""
Get the rows in the tables as list of messages.
The arguments are used for filtering.
"""
kwargs = locals().copy()
params = {}
for k,v in kwargs.items():
if k not in ('self',) and not v is None:
params[k] = v
rows = self._get_request('list', params)
return [
MessageDbRow.model_validate(
row, context=self.pydantic_context
) for row in rows
]
@validate_call
def total_messages(self,
id:str|None=None, sender:str|None=None, recipient:str|None=None,
processed:bool|None=None, solution:bool|None=None,
time_after:int|None=None, time_before:int|None=None
) -> int:
"""
Get the total number of rows in the tables matching the filters.
"""
kwargs = locals().copy()
params = {}
for k,v in kwargs.items():
if k not in ('self',) and not v is None:
params[k] = v
return int(self._get_request('app/table/total', params))
def _get_request(self, endpoint:str, params:Dict[str, Any]):
r = requests.get(
"{}/status".format(self.url),
params={"count": count}
"{}/{}".format(self.MANAGEMENT_URL, endpoint),
params=params
)
if r.status_code == 200:
return MessageDbRow.model_validate_json(r.text)
return r.json()
else:
raise RequestException(str(r.text)+str(r.headers))
raise RequestException(str(r.text)+"\n"+str(r.headers))
@validate_call
def send_message(self, ) -> AgentResponse:
# TODO
pass
def _post_request(self, message:AgentMessage) -> AgentResponse:
# TODO
def send_message(self, message:AgentMessage) -> AgentResponse:
r = requests.post(
"{}/message".format(self.url),
data=message.model_dump_json(),

8
ums/utils/types.py
View File

@ -90,7 +90,7 @@
"""
import os
import os, warnings
from enum import Enum
@ -108,6 +108,12 @@ from ums.utils.const import SHARE_PATH
from ums.utils.schema import ExtractionSchema
class RiddleInformation(BaseModel):
# ignore:
# /usr/local/lib/python3.12/dist-packages/pydantic/_internal/_fields.py:172:
# UserWarning: Field name "validate" in "RiddleStatus" shadows an attribute in parent
# "RiddleInformation"
warnings.filterwarnings('ignore', category=UserWarning, lineno=172, module="pydantic")
"""
This is the basic class used as superclass for all message and infos
about a riddle.