feat(remotes): full remote walkthrough, complete local remotes with editable functions
This commit is contained in:
1 parent
5009a4eb4c
commit
da1420d04b
4 files changed
+206
-48
No files matched your search
@@ -149,6 +149,12 @@ class App(ChatMixin):
|
||||
tools.append(RemoteToolSpec(connection, fn, name=prefix + fn.name))
|
||||
return tools
|
||||
|
||||
def resolve_remote_call(self, request_id, result=None, error=None):
|
||||
if result is not None:
|
||||
RemoteConnection.resolve(request_id, result);
|
||||
else:
|
||||
RemoteConnection.reject(request_id, error);
|
||||
|
||||
# --- Agent Methods (Asynchronous) ---
|
||||
|
||||
async def get_agent(self, agent_id):
|
||||
|
||||
+17
-12
@@ -39,7 +39,9 @@ def parse_function_specs(specs):
|
||||
)
|
||||
return functions
|
||||
|
||||
|
||||
# Global call dictionary
|
||||
# requestId -> asyncio.Future
|
||||
pending_calls = {}
|
||||
class RemoteConnection:
|
||||
"""A remote connected by a session via the 'remote-connect' action.
|
||||
|
||||
@@ -57,7 +59,6 @@ class RemoteConnection:
|
||||
self.name = name
|
||||
self.functions = functions # dict[str, RemoteFunction]
|
||||
self.send = send # async callable(dict) over the ws
|
||||
self.pending_calls = {} # requestId -> asyncio.Future
|
||||
self._request_ids = itertools.count()
|
||||
|
||||
def describe(self):
|
||||
@@ -88,11 +89,12 @@ class RemoteConnection:
|
||||
# be routed even when a session serves several remotes
|
||||
request_id = f"{self.name}#{next(self._request_ids)}"
|
||||
future = asyncio.get_running_loop().create_future()
|
||||
self.pending_calls[request_id] = future
|
||||
pending_calls[request_id] = future
|
||||
try:
|
||||
await self.send({
|
||||
"action": "remote-call",
|
||||
"requestId": request_id,
|
||||
"remotename": self.name,
|
||||
"function": function,
|
||||
"args": args,
|
||||
})
|
||||
@@ -101,32 +103,35 @@ class RemoteConnection:
|
||||
raise RemoteError(
|
||||
f"remote call '{function}' on '{self.name}' timed out after {timeout}s")
|
||||
finally:
|
||||
self.pending_calls.pop(request_id, None)
|
||||
pending_calls.pop(request_id, None)
|
||||
|
||||
def resolve(self, request_id, result):
|
||||
@classmethod
|
||||
def resolve(cls, request_id, result):
|
||||
"""Complete a pending call with the remote's result. No-op for
|
||||
unknown request ids (results are fanned out over a session's
|
||||
connections, see the remote-call-result handler)."""
|
||||
future = self.pending_calls.get(request_id)
|
||||
future = pending_calls.get(request_id)
|
||||
if future is not None and not future.done():
|
||||
future.set_result(result)
|
||||
else:
|
||||
raise Exception(f"remote request id '{request_id}' not found")
|
||||
|
||||
def reject(self, request_id, error):
|
||||
@classmethod
|
||||
def reject(cls, request_id, error):
|
||||
"""Fail a pending call with an error reported by the remote. No-op
|
||||
for unknown request ids."""
|
||||
future = self.pending_calls.get(request_id)
|
||||
future = pending_calls.get(request_id)
|
||||
if future is not None and not future.done():
|
||||
future.set_exception(RemoteError(str(error)))
|
||||
else:
|
||||
raise Exception(f"remote request id '{request_id}' not found")
|
||||
|
||||
def close(self):
|
||||
"""Fail all pending calls; called by App.disconnectRemote on manual
|
||||
disconnect and on session close. Callers waiting in call() get
|
||||
RemoteError."""
|
||||
self.send = None
|
||||
for future in self.pending_calls.values():
|
||||
if not future.done():
|
||||
future.set_exception(RemoteError(f"remote '{self.name}' disconnected"))
|
||||
self.pending_calls.clear()
|
||||
# TODO: Fail all pending calls from this remote. Is it necessary though? they will timeout.
|
||||
|
||||
|
||||
class RemoteToolSpec:
|
||||
|
||||
+12
-11
@@ -285,11 +285,17 @@ async def start_server(app, shutdown_event):
|
||||
# Invoke a function of one of the user's connected remotes.
|
||||
# Expected failures (unknown remote/function, timeout,
|
||||
# remote-side error) are replied, not raised.
|
||||
try:
|
||||
result = await app.callRemote(user_id, body['remotename'], body['function'], body.get('args', {}))
|
||||
await send({'result': result, 'nonce': nonce})
|
||||
except Exception as e:
|
||||
await send({'error': str(e), 'nonce': nonce})
|
||||
|
||||
|
||||
async def _handle_remote_call(user_id, remotename, func_name, args, send, nonce):
|
||||
try:
|
||||
result = await app.callRemote(user_id, remotename, func_name, args)
|
||||
await send({'result': result, 'nonce': nonce})
|
||||
except Exception as e:
|
||||
await send({'error': str(e), 'nonce': nonce})
|
||||
asyncio.create_task(_handle_remote_call(
|
||||
user_id, body['remotename'], body['function'], body.get('args', {}), send, nonce
|
||||
))
|
||||
|
||||
elif action == "remote-call-result":
|
||||
# A session serving remotes replies to a downstream call.
|
||||
@@ -297,12 +303,7 @@ async def start_server(app, shutdown_event):
|
||||
# requestIds are '<remotename>#<n>' and resolve/reject
|
||||
# are no-ops for unknown ids, so fanning out over this
|
||||
# session's connections routes the reply correctly.
|
||||
if 'error' in req:
|
||||
for conn in session_connections.values():
|
||||
conn.reject(req.get('requestId'), req['error'])
|
||||
else:
|
||||
for conn in session_connections.values():
|
||||
conn.resolve(req.get('requestId'), req.get('result'))
|
||||
app.resolve_remote_call(body['requestId'], body.get('result'), error=body.get('error'))
|
||||
|
||||
elif action == "create-chat":
|
||||
chat_id = await app.create_chat(
|
||||
|
||||
Reference in new issue
Block a user