Spaces:
Running
Running
File size: 3,336 Bytes
bf226b4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | from abc import abstractmethod
import json
import threading
from typing import Union, TypedDict, Dict, Any
from attr import dataclass
from flask import Request, Response, jsonify, Flask, session, request, send_file
from agent import AgentContext
from initialize import initialize_agent
from python.helpers.print_style import PrintStyle
from python.helpers.errors import format_error
from werkzeug.serving import make_server
Input = dict
Output = Union[Dict[str, Any], Response, TypedDict] # type: ignore
class ApiHandler:
def __init__(self, app: Flask, thread_lock: threading.Lock):
self.app = app
self.thread_lock = thread_lock
@classmethod
def requires_loopback(cls) -> bool:
return False
@classmethod
def requires_api_key(cls) -> bool:
return False
@classmethod
def requires_auth(cls) -> bool:
return True
@classmethod
def get_methods(cls) -> list[str]:
return ["POST"]
@classmethod
def requires_csrf(cls) -> bool:
return cls.requires_auth()
@abstractmethod
async def process(self, input: Input, request: Request) -> Output:
pass
async def handle_request(self, request: Request) -> Response:
try:
# input data from request based on type
input_data: Input = {}
if request.is_json:
try:
if request.data: # Check if there's any data
input_data = request.get_json()
# If empty or not valid JSON, use empty dict
except Exception as e:
# Just log the error and continue with empty input
PrintStyle().print(f"Error parsing JSON: {str(e)}")
input_data = {}
else:
input_data = {"data": request.get_data(as_text=True)}
# process via handler
output = await self.process(input_data, request)
# return output based on type
if isinstance(output, Response):
return output
else:
response_json = json.dumps(output)
return Response(
response=response_json, status=200, mimetype="application/json"
)
# return exceptions with 500
except Exception as e:
error = format_error(e)
PrintStyle.error(f"API error: {error}")
return Response(response=error, status=500, mimetype="text/plain")
# get context to run agent zero in
#
# PAPERCLIP FIX (2026-07-01): when ctxid is empty, this used to silently
# reuse AgentContext.first() -- meaning EVERY caller that omitted the
# context id (e.g. a fresh session from an external API client) landed
# on the exact same shared conversation/context as every other caller
# that also omitted it. That defeated per-caller session isolation.
# Now an empty ctxid always creates a brand-new, isolated context.
def get_context(self, ctxid: str):
with self.thread_lock:
if not ctxid:
return AgentContext(config=initialize_agent())
got = AgentContext.get(ctxid)
if got:
return got
return AgentContext(config=initialize_agent(), id=ctxid)
|