xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-06-04 21:06:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: query-based apis

Browse Source
This commit is contained in:
Junyan Qin
2025-07-13 18:41:04 +08:00
parent 9f269d1614
commit 4a319b2b20
3 changed files with 59 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
pkg/plugin/handler.py
View File

@@ -107,6 +107,62 @@ class RuntimeConnectionHandler(handler.Handler):
},
)
@self.action(PluginToRuntimeAction.SET_QUERY_VAR)
async def set_query_var(data: dict[str, Any]) -> handler.ActionResponse:
"""Set query var"""
query_id = data['query_id']
key = data['key']
value = data['value']
if query_id not in self.ap.query_pool.cached_queries:
return handler.ActionResponse.error(
message=f'Query with query_id {query_id} not found',
)
query = self.ap.query_pool.cached_queries[query_id]
query.variables[key] = value
return handler.ActionResponse.success(
data={},
)
@self.action(PluginToRuntimeAction.GET_QUERY_VAR)
async def get_query_var(data: dict[str, Any]) -> handler.ActionResponse:
"""Get query var"""
query_id = data['query_id']
key = data['key']
if query_id not in self.ap.query_pool.cached_queries:
return handler.ActionResponse.error(
message=f'Query with query_id {query_id} not found',
)
query = self.ap.query_pool.cached_queries[query_id]
return handler.ActionResponse.success(
data={
'value': query.variables[key],
},
)
@self.action(PluginToRuntimeAction.GET_QUERY_VARS)
async def get_query_vars(data: dict[str, Any]) -> handler.ActionResponse:
"""Get query vars"""
query_id = data['query_id']
if query_id not in self.ap.query_pool.cached_queries:
return handler.ActionResponse.error(
message=f'Query with query_id {query_id} not found',
)
query = self.ap.query_pool.cached_queries[query_id]
return handler.ActionResponse.success(
data={
'vars': query.variables,
},
)
async def ping(self) -> dict[str, Any]:
"""Ping the runtime"""
return await self.call_action(