初始化对话历史# 加载系统提示词system_promptself._load_prompt()# 对话历史零初始化对话历史文messages: list[dict][{role:system,content:system_prompt},{role:user,content:(f请对以下已授权目标开展测试{target}\nf授权范围{, .join(self.settings.allowed_targets)}\n请按预设流程先理解目标再选择合适的工具取证最后输出有真实证据支持的测试报告。),},]system 消息只初始化一次。{“role0”: “system”, “content”: system_prompt},{“role0”: “user”, “content”: f请对以下已授权目标开展测试…}进入循环第一轮tool_choice强制选择http_checkforce_toolTrue tool_choice1{type:function,function:{name:http_check}}# 发送瞬间历史只有两条[system,user]# tool_choice1强制模型必须返回http_check的tool_callllm.chat(messages, tool_schemas,tool_choicetool_choice1)# 模型返回后将带有http_check的tool_call进行appendmessages.append(self._assistant_message(response))此时{“role0”: “system”, “content”: system_prompt},{“role0”: “user”, “content”: f请对以下已授权目标开展测试…“}{“role”:“assistant”,“content”:null,“tool_calls”:[{“id”:“call_001”,“type”:“function”,“function”:{“name”:“http_check”,“arguments”:”{“url”:“http://192.168.1.100”}}}]}经过response.tool_calls for循环messages.append({role:tool,tool_call_id:tc.id,content:tool_content}){“role0”: “system”, “content”: system_prompt},{“role0”: “user”, “content”: f请对以下已授权目标开展测试…“}{“role”:“assistant”,“content”:null,“tool_calls”:[{“id”:“call_001”,“type”:“function”,“function”:{“name”:“http_check”,“arguments”:”{“url”:“http://192.168.1.100”}}}]}{“role”:“tool”,“tool_call_id”:“call_001”,“content”:“HTTP状态码200服务器nginx响应长度1024”}第二轮tool_choice可以autoforce_toolFalse tool_choice2{type:function,function:{name:auto}}# 发送瞬间历史只有四条条[system,user]llm.chat(messages, tool_schemas,tool_choicetool_choice2)# 模型返回后将带有auto的tool_call进行appendmessages.append(self._assistant_message(response))完整代码defrun(self,target:str)-AgentRunResult:invocations:list[ToolInvocation][]stopped_reason未知final_texttool_schemasself.registry.openai_tools()forturninrange(1,self.settings.max_agent_turns1):logger.info( 第 %d/%d 轮决策 ,turn,self.settings.max_agent_turns)# 强制调用工具force_toollen(invocations)0# tool_choice是三元表达式如果force_tool(默认0 False)走前强制http_check工具# 调用过则force_tool为True走自动选择工具tool_choice({type:function,function:{name:http_check}}ifforce_toolelseauto)try:# system# 假设使用了DeepSeek模型 self.llm就是DeepSeek客户端.chat调用聊天接口response:LLMResponseself.llm.chat(# messages是对话上下文tool_schemas告诉模型有哪些工具可用# tool_choice是deep选中的工具messages,tool_schemas,tool_choicetool_choice)exceptLLMErrorasexc:# 模型错误记录并结束不伪造报告。logger.error(模型调用失败终止任务%s,exc)stopped_reasonf模型错误{exc}final_textself._fallback_report(target,invocations,stopped_reason)break# 这轮对话 记入对话历史messages.append(self._assistant_message(response))# 没有工具调用ifnotresponse.has_tool_calls:ifforce_tool:# 被要求取证却仍输出纯文本该文本没有真实执行依据不采信# 追加纠正消息后继续直到模型真正调用工具。logger.warning(模型未调用工具就输出结果已纠正并要求先取证)messages.append({role:user,content:(你还没有调用任何工具上面的状态码、响应大小、耗时等都没有真实执行依据属于无效内容。请立即调用 http_check 工具对目标取证禁止在工具返回真实结果前输出任何数字或最终报告。),})continue# 已有真实工具结果模型输出最终报告 - 结束循环。final_textresponse.textor(模型未返回文本)stopped_reason模型结束输出最终报告logger.info(模型不再调用工具任务结束)break# 有工具调用逐个校验并在容器内真实执行再把结果回传。fortcinresponse.tool_calls:logger.info(模型请求工具%s 参数%s,tc.name,tc.arguments)try:tool_resultself.registry.execute(tc.name,tc.arguments,self.runtime)invocationToolInvocation(turnturn,nametc.name,argumentstc.arguments,oktool_result.ok,result_summarytool_result.summary,datatool_result.data,errortool_result.error,)tool_contenttool_result.to_message()exceptToolErrorasexc:# 校验/执行失败也要作为 tool 消息回传让模型据此修正。logger.warning(工具调用被拒绝/失败%s,exc)invocationToolInvocation(turnturn,nametc.name,argumentstc.arguments,okFalse,result_summary,errorstr(exc),)tool_content(f[工具{tc.name}执行失败]\n{exc}\n请修正参数或选择其它操作不要重复同样的错误请求。)invocations.append(invocation)# tool 角色消息必须带对应 tool_call_idmessages.append({role:tool,tool_call_id:tc.id,content:tool_content})else:# 达到轮数上限仍未结束基于已有真实结果生成兜底报告。stopped_reasonf达到最大轮数{self.settings.max_agent_turns}logger.warning(stopped_reason)final_textself._fallback_report(target,invocations,stopped_reason)elapsedtime.monotonic()-started logger.info(循环结束%s共 %d 次工具调用耗时 %.1fs,stopped_reason,len(invocations),elapsed)report_path,trace_pathself._persist(target,final_text,messages,invocations,stopped_reason,elapsed)returnAgentRunResult(targettarget,report_textfinal_text,report_pathreport_path,trace_pathtrace_path,turns_usedmin(turn,self.settings.max_agent_turns),stopped_reasonstopped_reason,invocationsinvocations,)MCP因为已经在registry.py中注册了registry.register(CALL_MCP_SPEC)当self.registry.execute() 接收到 tc.name “call_mcp” 时直接找到CALL_MCP_SPEC实例CALL_MCP_SPECToolSpec(namecall_mcp,description(调用指定 MCP 连接上的一个工具并返回真实结果。先用 list_mcps 确认连接名与工具名再按该工具的 input schema 传 arguments无参数工具可省略。),parameters{type:object,properties:{connection:{type:string,description:MCP 连接名与 list_mcps 返回的 name 一致,},tool:{type:string,description:目标连接上的工具名与 list_mcps 返回的工具名一致,},arguments:{type:object,description:(工具参数键值对形式形状匹配该工具的 input_schema无参数工具可省略),additionalProperties:True,default:{},},},required:[connection,tool],additionalProperties:False,},handler_call_mcp,)通过 CALL_MCP_SPEC.handler 调用 _call_mcp 函数def _call_mcp(arguments: dict[str, Any], runtime: ToolRuntime)-ToolResult: 从参数中提取MCP Server名称、工具名、工具参数 connectionfilesystemtoolechotool_arguments{message:hello world} connectionarguments.get(connection)toolarguments.get(tool)tool_argumentsarguments.get(arguments,{})or{}# 必须非空字符串ifnot isinstance(connection, str)or not connection.strip(): raise ToolValidationError(connection 必须是非空字符串)ifnot isinstance(tool, str)or not tool.strip(): raise ToolValidationError(tool 必须是非空字符串)ifnot isinstance(tool_arguments, dict): raise ToolValidationError(arguments 必须是 JSON 对象)# 获取 MCP 配置文件获取 MCP Server 连接信息configsload_mcp_configs()try:# 调用client.call_tool方法调用MCP Server上的工具# 传入4参数 configs echo {message: hello world}payloadclient.call_tool(configs, connection.strip(), tool.strip(), tool_arguments)except ValueError as exc:# 连接/工具不存在、白名单拒绝等作为失败结果回传给模型不抛断循环returnToolResult(namecall_mcp,okFalse,summary,data{connection:connection,tool:tool},errorstr(exc),)ifpayload[is_error]:returnToolResult(namecall_mcp,okFalse,summary,datapayload,error(MCP 服务器返回工具错误f{payload[text] or (无错误文本)}),)# 提取工具返回内容textpayload[text]or(工具未返回文本内容)ifpayload[structured]: text(f\n[另有 {len(payload[structured])} 项非文本结构化内容见 data.structured])returnToolResult(namecall_mcp,okTrue,summaryf连接 {connection} 的工具 {tool} 调用成功\n{text},datapayload,)最后return ToolResult(name“call_mcp”,okTrue,summaryf连接 {connection} 的工具 {tool} 调用成功\n{text},datapayload,)与MCP进行连接
