(self, request: Request)
| 60 | pass |
| 61 | |
| 62 | async def handle_request(self, request: Request) -> Response: |
| 63 | try: |
| 64 | # input data from request based on type |
| 65 | input_data: Input = {} |
| 66 | if request.is_json: |
| 67 | try: |
| 68 | if request.data: # Check if there's any data |
| 69 | input_data = request.get_json() |
| 70 | # If empty or not valid JSON, use empty dict |
| 71 | except Exception as e: |
| 72 | # Just log the error and continue with empty input |
| 73 | PrintStyle().print(f"Error parsing JSON: {str(e)}") |
| 74 | input_data = {} |
| 75 | else: |
| 76 | # input_data = {"data": request.get_data(as_text=True)} |
| 77 | input_data = {} |
| 78 | |
| 79 | # process via handler |
| 80 | output = await self.process(input_data, request) |
| 81 | |
| 82 | # return output based on type |
| 83 | if isinstance(output, Response): |
| 84 | return output |
| 85 | else: |
| 86 | response_json = json.dumps(output) |
| 87 | return Response( |
| 88 | response=response_json, status=200, mimetype="application/json" |
| 89 | ) |
| 90 | |
| 91 | # return exceptions with 500 |
| 92 | except Exception as e: |
| 93 | error = format_error(e) |
| 94 | PrintStyle.error(f"API error: {error}") |
| 95 | return Response(response=error, status=500, mimetype="text/plain") |
| 96 | |
| 97 | # get context to run agent zero in |
| 98 | def use_context(self, ctxid: str, create_if_not_exists: bool = True): |
no test coverage detected