defmodule CMDCGateway.CallbackTool do @moduledoc """ HTTP 回调工具代理。 外部系统通过 `POST /v1/sessions/:id/tools` 注册自定义工具, Agent 调用时 Gateway 向 `callback_url` 发 POST 请求,代理执行结果。 ## 注册请求体 { "name": "query_database", "description": "Run a SQL query on the production database", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "SQL query to execute"} }, "required": ["query"] }, "callbackUrl": "https://my-service.example.com/tools/query", "timeoutMs": 30000 } ## 回调请求格式 Gateway 向 callbackUrl 发 POST: { "callId": "tc_abc123", "toolName": "query_database", "args": {"query": "SELECT * FROM users LIMIT 10"}, "sessionId": "s-001" } ## 回调响应格式 {"result": "10 rows returned..."} // 或错误 {"error": "Connection refused"} """ @registry_table :"#{__MODULE__}.Registry" @type tool_spec :: %{ name: String.t(), description: String.t(), parameters: map(), callback_url: String.t(), timeout_ms: pos_integer() } # ========================================================================== # Public API # ========================================================================== @doc "初始化 ETS 注册表(由 Application 调用)。" @spec init_registry() :: :ok def init_registry do if :ets.info(@registry_table) == :undefined do :ets.new(@registry_table, [:named_table, :bag, :public, read_concurrency: true]) end :ok end @doc """ 为指定 session 注册一个回调工具。 返回动态生成的 Tool 模块原子。 """ @spec register(String.t(), map()) :: {:ok, module()} | {:error, term()} def register(session_id, %{"name" => name, "callbackUrl" => callback_url} = spec) do tool_spec = %{ name: name, description: Map.get(spec, "description", "External tool: #{name}"), parameters: Map.get(spec, "parameters", %{"type" => "object", "properties" => %{}}), callback_url: callback_url, timeout_ms: Map.get(spec, "timeoutMs", 30_000) } tool_module = generate_tool_module(session_id, tool_spec) :ets.insert(@registry_table, {session_id, name, tool_module, tool_spec}) {:ok, tool_module} end def register(_session_id, _spec), do: {:error, "Missing required fields: name, callbackUrl"} @doc "获取指定 session 的所有已注册回调工具模块。" @spec list_tools(String.t()) :: [module()] def list_tools(session_id) do if :ets.info(@registry_table) != :undefined do @registry_table |> :ets.lookup(session_id) |> Enum.map(fn {_sid, _name, module, _spec} -> module end) else [] end end @doc "清理指定 session 的所有回调工具。" @spec cleanup(String.t()) :: :ok def cleanup(session_id) do if :ets.info(@registry_table) != :undefined do :ets.match_delete(@registry_table, {session_id, :_, :_, :_}) end :ok end @doc """ 执行 HTTP 回调。 向 `callback_url` 发送 POST 请求,传递工具调用参数,返回结果。 """ @spec execute_callback(tool_spec(), String.t(), String.t(), map()) :: {:ok, String.t()} | {:error, String.t()} def execute_callback(tool_spec, session_id, call_id, args) do body = %{ callId: call_id, toolName: tool_spec.name, args: args, sessionId: session_id } tool_spec.callback_url |> Req.post(json: body, receive_timeout: tool_spec.timeout_ms, retry: false) |> handle_callback_response() end defp handle_callback_response({:ok, %Req.Response{status: status, body: body}}) when status in 200..299 do body |> extract_result() |> normalize_result() end defp handle_callback_response({:ok, %Req.Response{status: status, body: body}}) do {:error, "Callback returned HTTP #{status}: #{inspect(body)}"} end defp handle_callback_response({:error, reason}) do {:error, "Callback request failed: #{inspect(reason)}"} end defp extract_result(%{"result" => r}), do: r defp extract_result(%{"error" => e}), do: {:error, e} defp extract_result(r) when is_binary(r), do: r defp extract_result(r), do: inspect(r) defp normalize_result({:error, e}), do: {:error, to_string(e)} defp normalize_result(r), do: {:ok, to_string(r)} # ========================================================================== # Private — Dynamic Tool Module Generation # ========================================================================== defp generate_tool_module(session_id, tool_spec) do module_name = Module.concat([ CMDCGateway.CallbackTool.Dynamic, "S_#{safe_name(session_id)}_T_#{safe_name(tool_spec.name)}" ]) if :erlang.module_loaded(module_name) do :code.purge(module_name) :code.delete(module_name) end spec = tool_spec contents = quote do @moduledoc false @behaviour CMDC.Tool @tool_spec unquote(Macro.escape(spec)) @session_id unquote(session_id) @impl true def name, do: @tool_spec.name @impl true def description, do: @tool_spec.description @impl true def parameters, do: @tool_spec.parameters @impl true def execute(args, ctx) do call_id = Map.get(ctx, :call_id, "unknown") sid = Map.get(ctx, :session_id, @session_id) CMDCGateway.CallbackTool.execute_callback(@tool_spec, sid, call_id, args) end end Module.create(module_name, contents, Macro.Env.location(__ENV__)) module_name end defp safe_name(str) do str |> String.replace(~r/[^a-zA-Z0-9]/, "_") |> String.slice(0, 40) end end