Aurai Advisor (上级顾问 MCP)

Instructions

# 步骤 1：将代码文件复制为 .txt
shutil.copy('script.py', 'script.txt')

# 步骤 2：上传文件
sync_context(operation='incremental', files=['script.txt'])

# 步骤 3：告诉上级顾问文件已上传
consult_aurai(
    error_message='请审查已上传的 script.txt 文件'
)

{
  "status": "need_info",
  "questions_to_answer": ["问题1", "问题2"],
  "instruction": "请搜集信息并再次调用"
}

{
  "status": "success",
  "analysis": "问题分析",
  "guidance": "解决建议",
  "action_items": ["步骤1", "步骤2"],
  "resolved": false  // 是否已完全解决
}

# 第一次咨询（问题A）
consult_aurai(problem_type="runtime_error", error_message="...")

# 继续讨论问题A...
consult_aurai(answers_to_questions="...")

# 切换到问题B（标注为新问题，清空历史）
consult_aurai(
    problem_type="design_issue",
    error_message="...",
    is_new_question=True  # [注意] 会清空之前关于问题A的所有对话
)

Implementation Reference

• src/mcp_aurai/server.py:121-386 (handler)

  Main handler function for consult_aurai tool - async function that processes consultation requests, manages conversation history, builds prompts, calls the AI client, and returns guidance or follow-up questions based on the AI's response

  async def consult_aurai(
      problem_type: str = Field(
          description="问题类型: runtime_error, syntax_error, design_issue, other"
      ),
      error_message: str = Field(description="错误描述"),
      code_snippet: str | None = Field(default=None, description="相关代码片段"),
      context: Any = Field(default=None, description="上下文信息（支持 JSON 字符串或字典，会自动解析）"),
      attempts_made: str | None = Field(default=None, description="已尝试的解决方案"),
      answers_to_questions: str | None = Field(
          default=None,
          description="对上级顾问反问的回答（仅在多轮对话时使用）"
      ),
      is_new_question: bool = Field(
          default=False,
          description="[重要] 是否为新问题（新问题会清空之前的所有对话历史，确保干净的上下文）"
      ),
  ) -> dict[str, Any]:
      """
      请求上级AI的指导（支持交互对齐机制与多轮对话）
  
      这是核心工具，当本地AI遇到编程问题时调用此工具获取上级AI的指导建议。
  
      ---
  
      **🔗 相关工具**
  
      - **sync_context**：需要上传文档或代码时使用
        - 📄 上传文章、说明文档（.md/.txt）
        - 💻 **上传代码文件（避免内容被截断）** ⭐ 重要
        - 将 `.py/.js/.json` 等代码文件复制为 `.txt` 后上传
  
      - **report_progress**：执行上级 AI 建议后，使用此工具报告进度并获取下一步指导
  
      - **get_status**：查看当前对话状态、迭代次数、配置信息
  
      **💡 重要提示：避免内容被截断**
  
      如果 `code_snippet` 或 `context` 内容过长，**请使用 `sync_context` 上传文件**：
  
      ```python
      # 步骤 1：将代码文件复制为 .txt
      shutil.copy('script.py', 'script.txt')
  
      # 步骤 2：上传文件
      sync_context(operation='incremental', files=['script.txt'])
  
      # 步骤 3：告诉上级顾问文件已上传
      consult_aurai(
          error_message='请审查已上传的 script.txt 文件'
      )
      ```
  
      **优势**：
      - ✅ 避免代码在 `context` 或 `answers_to_questions` 字段中被截断
      - ✅ 利用文件读取机制，完整传递内容
      - ✅ 支持任意大小的代码文件
  
      ---
  
      ## [重要] 何时开始新对话？
  
      **系统会自动检测**，但你也可以手动控制：
  
      - **自动清空**：当上一次对话返回 `resolved=true` 时，系统会自动清空历史
      - **手动清空**：如果你要讨论一个完全不同的新问题，设置 `is_new_question=true`
  
      **何时设置 `is_new_question=true`？**
      - [OK] 切换到完全不相关的项目/文件
      - [OK] 之前的问题已解决，现在遇到全新的问题
      - [OK] 发现上下文混乱，想重新开始
      - [X] 不要在同一个问题的多轮对话中使用
  
      ## 交互协议
  
      ### 1. 多轮对齐机制
      - **不要期待一次成功**：上级顾问可能会认为信息不足，返回反问问题
      - 仔细阅读 `questions_to_answer` 中的每个问题
      - 主动搜集信息（读取文件、检查日志、运行命令）
      - **再次调用** 此工具，将答案填入 `answers_to_questions` 参数
  
      ### 2. 首次调用
      必须提供：
      - `problem_type`：问题类型（runtime_error/syntax_error/design_issue/other）
      - `error_message`：清晰描述问题或错误
      - `context`：相关上下文（代码片段、环境信息、已尝试的方案）
      - `code_snippet`：相关代码（如果有）
  
      ### 3. 后续调用（当返回 status="need_info" 时）
      必须提供：
      - `answers_to_questions`：对上级顾问反问的详细回答
      - 保持其他参数不变（除非有新信息）
  
      ### 4. 诚实原则
      - **禁止瞎编**：如果不知道答案，诚实说明"未找到相关信息"
      - **禁止臆测**：不要在没有证据的情况下假设解决方案
      - 提供具体证据（文件路径、日志内容、错误堆栈）
  
      ## 响应格式
  
      ### 信息不足时 (status="need_info")
      ```json
      {
        "status": "need_info",
        "questions_to_answer": ["问题1", "问题2"],
        "instruction": "请搜集信息并再次调用"
      }
      ```
  
      ### 提供指导时 (status="success")
      ```json
      {
        "status": "success",
        "analysis": "问题分析",
        "guidance": "解决建议",
        "action_items": ["步骤1", "步骤2"],
        "resolved": false  // 是否已完全解决
      }
      ```
  
      ### 问题解决后
      当 `resolved=true` 时，对话历史会自动清空，下次查询将开始新对话。
  
      ### [自动] 新对话检测
      系统会自动检测新问题：
      - 如果上一次对话的 `resolved=true`，下次调用 `consult_aurai` 时会自动清空历史
      - 保证每个独立问题都有干净的上下文，避免干扰
  
      ### [重要] 明确标注新问题（可选参数）
      如果你想强制开始一个新对话，可以设置 `is_new_question=true`：
      - **效果**：立即清空所有之前的对话历史
      - **后果**：上级AI将无法看到之前的任何上下文
      - **使用场景**：
        - 之前的对话已完全无关
        - 想重新开始讨论一个全新的问题
        - 发现上下文混乱，想重置
  
      **示例**：
      ```python
      # 第一次咨询（问题A）
      consult_aurai(problem_type="runtime_error", error_message="...")
  
      # 继续讨论问题A...
      consult_aurai(answers_to_questions="...")
  
      # 切换到问题B（标注为新问题，清空历史）
      consult_aurai(
          problem_type="design_issue",
          error_message="...",
          is_new_question=True  # [注意] 会清空之前关于问题A的所有对话
      )
      ```
      """
      config = get_aurai_config()
  
      logger.info(f"收到consult_aurai请求，问题类型: {problem_type}，是否新问题: {is_new_question}")
  
      # [新问题] 处理新问题：两种方式触发清空历史
      # 方式1：明确标注 is_new_question=true
      # 方式2：自动检测（上一次对话已解决）
      should_clear_history = False
      clear_reason = ""
  
      if is_new_question:
          # 明确标注新问题
          should_clear_history = True
          clear_reason = "下级AI明确标注为新问题"
      elif _conversation_history:
          # 自动检测：检查上一次对话是否已解决
          last_entry = _conversation_history[-1]
          last_response = last_entry.get("response", {})
  
          if last_response.get("resolved", False):
              should_clear_history = True
              clear_reason = "上一次对话已解决（自动检测）"
  
      # 执行清空操作
      if should_clear_history:
          history_count = len(_conversation_history)
          _conversation_history.clear()
          logger.info(f"[新问题] 清空对话历史（清除 {history_count} 条记录）")
          logger.info(f"   原因: {clear_reason}")
          logger.info(f"   新问题: {problem_type} - {error_message[:100]}...")
  
      # 解析 context 参数（支持 JSON 字符串或字典）
      parsed_context: dict[str, Any] = {}
      if context:
          if isinstance(context, str):
              try:
                  parsed_context = json.loads(context)
                  logger.debug("已解析 JSON 格式的 context")
              except json.JSONDecodeError as e:
                  logger.warning(f"context JSON 解析失败: {e}，使用空字典")
                  parsed_context = {}
          elif isinstance(context, dict):
              parsed_context = context
  
      # 构建提示词（如果有对反问的回答，加入上下文）
      current_context = parsed_context or {}
      if answers_to_questions:
          current_context["answers_to_questions"] = answers_to_questions
  
      prompt = build_consult_prompt(
          problem_type=problem_type,
          error_message=error_message,
          code_snippet=code_snippet,
          context=current_context,
          attempts_made=attempts_made,
          iteration=len(_conversation_history),
          conversation_history=_get_history(),
      )
  
      # 调用上级AI，传递对话历史
      client = get_aurai_client()
      response = await client.chat(
          user_message=prompt,
          conversation_history=_get_history()
      )
  
      # 记录到历史
      _add_to_history({
          "type": "consult",
          "problem_type": problem_type,
          "error_message": error_message,
          "response": response,
          "had_answers": answers_to_questions is not None,
      })
  
      # 根据上级顾问的响应状态返回不同格式
      if response.get("status") == "aligning":
          # 模式 A: 信息不足，需要补充
          logger.info(f"上级顾问要求补充信息，问题数: {len(response.get('questions', []))}")
          return {
              "status": "need_info",
              "message": "[提示] 上级顾问认为信息不足，请回答以下问题：",
              "questions_to_answer": response.get("questions", []),
              "instruction": "请搜集信息，再次调用 consult_aurai，并将答案填入 'answers_to_questions' 字段。",
              # ⭐ 相关工具提示
              "related_tools_hint": {
                  "sync_context": {
                      "description": "如果需要上传文档（.md/.txt）来补充上下文信息",
                      "example": "sync_context(operation='full_sync', files=['path/to/doc.md'])"
                  }
              }
          }
      else:
          # 模式 B: 信息充足，提供指导
          logger.info(f"上级顾问提供指导，resolved: {response.get('resolved', False)}")
  
          # 检查问题是否已解决，若解决则清空对话历史
          if response.get("resolved", False):
              history_count = len(_conversation_history)
              _conversation_history.clear()
              logger.info(f"[完成] 问题已解决，已清空对话历史（清除了 {history_count} 条记录）")
  
          return {
              "status": "success",
              "analysis": response.get("analysis"),
              "guidance": response.get("guidance"),
              "action_items": response.get("action_items", []),
              "code_changes": response.get("code_changes", []),
              "verification": response.get("verification"),
              "needs_another_iteration": response.get("needs_another_iteration", False),
              "resolved": response.get("resolved", False),
              "requires_human_intervention": response.get("requires_human_intervention", False),
              "hint": "[提示] 如需咨询新问题，下次调用时设置 is_new_question=true。这将清空之前的所有对话历史（包括之前的问题和上级AI的指导），但当前这条新问题会正常处理并保留在新的对话中",
          }

• src/mcp_aurai/server.py:120-120 (registration)

  Tool registration using @mcp.tool() decorator from FastMCP framework - registers consult_aurai as an MCP tool

  @mcp.tool()

• src/mcp_aurai/server.py:122-137 (schema)

  Input schema definition using Field validators - defines parameters (problem_type, error_message, code_snippet, context, attempts_made, answers_to_questions, is_new_question) with descriptions and validation

      problem_type: str = Field(
          description="问题类型: runtime_error, syntax_error, design_issue, other"
      ),
      error_message: str = Field(description="错误描述"),
      code_snippet: str | None = Field(default=None, description="相关代码片段"),
      context: Any = Field(default=None, description="上下文信息（支持 JSON 字符串或字典，会自动解析）"),
      attempts_made: str | None = Field(default=None, description="已尝试的解决方案"),
      answers_to_questions: str | None = Field(
          default=None,
          description="对上级顾问反问的回答（仅在多轮对话时使用）"
      ),
      is_new_question: bool = Field(
          default=False,
          description="[重要] 是否为新问题（新问题会清空之前的所有对话历史，确保干净的上下文）"
      ),
  ) -> dict[str, Any]:

• src/mcp_aurai/prompts.py:6-133 (helper)

  build_consult_prompt helper function - constructs the prompt sent to the upper-level AI advisor, including problem information, context, code snippets, conversation history, and response format instructions

  def build_consult_prompt(
      problem_type: str,
      error_message: str,
      code_snippet: str | None = None,
      context: dict[str, Any] | None = None,
      attempts_made: str | None = None,
      iteration: int = 0,
      conversation_history: list[dict[str, str]] | None = None,
  ) -> str:
      """
      构建请求上级AI指导的提示词
  
      Args:
          problem_type: 问题类型
          error_message: 错误描述
          code_snippet: 相关代码片段
          context: 上下文信息
          attempts_made: 已尝试的解决方案
          iteration: 当前迭代次数
          conversation_history: 对话历史
      """
      context = context or {}
  
      # 构建上下文描述
      context_desc = []
      if file_path := context.get("file_path"):
          context_desc.append(f"- 文件路径: {file_path}")
      if line_number := context.get("line_number"):
          context_desc.append(f"- 行号: {line_number}")
      if terminal_output := context.get("terminal_output"):
          context_desc.append(f"- 终端输出:\n```\n{terminal_output}\n```")
  
      # 构建对话历史
      history_desc = ""
      if conversation_history:
          history_desc = "\n## 对话历史\n\n"
          for i, turn in enumerate(conversation_history[-5:], 1):  # 只保留最近5轮
              history_desc += f"### 第{i}轮\n"
              if "action" in turn:
                  history_desc += f"**执行操作**: {turn['action']}\n"
              if "result" in turn:
                  history_desc += f"**执行结果**: {turn['result']}\n"
              history_desc += "\n"
  
      prompt = f"""# 你是上级AI顾问
  
  ## 角色
  你是一位经验丰富的技术顾问，正在指导一位"本地AI助手"解决编程问题。
  
  ## 任务
  分析以下问题，提供清晰、可执行的指导。
  
  ## 问题信息
  - **问题类型**: {problem_type}
  - **错误描述**: {error_message}
  - **当前迭代**: 第 {iteration + 1} 轮
  
  ## 上下文
  {chr(10).join(context_desc) if context_desc else "无"}
  
  ## 代码片段
  {f"```{context.get('language', 'python')}\n{code_snippet}\n```" if code_snippet else "无"}
  
  ## 已尝试方案
  {attempts_made if attempts_made else "无"}
  {history_desc}
  
  ## 你的回应格式
  
  **重要**: 首先评估信息完整度，然后选择对应的模式返回。
  
  **信息不足时（缺少具体错误、代码、上下文）**:
  ```json
  {{
    "status": "aligning",
    "questions": ["需补充的问题1", "需补充的问题2"],
    "analysis": null,
    "guidance": null
  }}
  ```
  
  **信息充足时**:
  ```json
  {{
    "status": "guiding",
    "questions": [],
    "analysis": "问题分析 - 简明扼要分析问题根源",
    "guidance": "指导建议 - 具体建议的文字描述",
    "action_items": ["步骤1", "步骤2", "..."],
    "code_changes": [
      {{
        "file": "文件路径",
        "line": 行号,
        "old": "原代码",
        "new": "新代码"
      }}
    ],
    "verification": "验证方法",
    "needs_another_iteration": false,
    "resolved": false,
    "requires_human_intervention": false
  }}
  ```
  
  ## 字段说明
  
  - **status**: 必须是 "aligning"（信息不足）或 "guiding"（信息充足）
  - **questions**: 反问列表（仅在 aligning 模式）
  - **analysis**: 分析问题根本原因
  - **guidance**: 给出具体、可执行的指导建议
  - **action_items**: 数组，列出具体的执行步骤
  - **code_changes**: 数组（可选），如果需要修改代码，列出具体的代码变更
  - **verification**: 验证方法（新增）
  - **needs_another_iteration**: 布尔值，是否需要继续迭代
  - **resolved**: 布尔值，问题是否已解决
  - **requires_human_intervention**: 布尔值，是否需要人工介入
  
  ## 重要原则
  
  1. **指导要具体、可执行** - 避免模糊建议
  2. **最小改动原则** - 优先考虑最小改动来解决问题
  3. **分步骤指导** - 如果问题复杂，分步骤给出建议
  4. **承认限制** - 承认无法解决的问题，及时设置 requires_human_intervention=true
  5. **避免无限循环** - 如果连续多次建议相同方向，考虑人工介入
  
  现在，请分析上述问题并给出你的指导。
  """
      return prompt

• src/mcp_aurai/llm.py:176-258 (helper)

  AuraiClient.chat async method - handles communication with the OpenAI-compatible API, sends messages with conversation history, parses JSON responses, and handles errors

  async def chat(
      self,
      user_message: str,
      system_prompt: str | None = None,
      response_format: Literal["text", "json_object"] = "json_object",
      conversation_history: list[dict] | None = None,
  ) -> dict:
      """
      发送聊天请求
  
      Args:
          user_message: 用户消息
          system_prompt: 系统提示词
          response_format: 响应格式
          conversation_history: 对话历史（用于多轮对话）
  
      Returns:
          解析后的JSON响应
      """
      from .prompts import SYSTEM_PROMPT
  
      system_prompt = system_prompt or SYSTEM_PROMPT
  
      messages = []
      if system_prompt:
          messages.append({"role": "system", "content": system_prompt})
  
      # 添加对话历史
      history_messages = self._build_messages_from_history(conversation_history)
      messages.extend(history_messages)
  
      # 添加当前用户消息
      messages.append({"role": "user", "content": user_message})
  
      logger.info(f"发送请求到 {self.config.base_url}，消息数: {len(messages)}")
  
      try:
          response = self._client.chat.completions.create(
              model=self.config.model,
              messages=messages,
              temperature=self.config.temperature,
              max_tokens=self.config.max_tokens,
          )
  
          content = response.choices[0].message.content
          logger.info(f"收到响应，长度: {len(content)}")
  
          # 尝试解析JSON
          try:
              # 清理可能存在的markdown代码块标记
              content_clean = content.strip()
              if content_clean.startswith("```json"):
                  content_clean = content_clean[7:]
              if content_clean.startswith("```"):
                  content_clean = content_clean[3:]
              if content_clean.endswith("```"):
                  content_clean = content_clean[:-3]
              content_clean = content_clean.strip()
  
              result = json.loads(content_clean)
              logger.info("成功解析JSON响应")
              return result
          except json.JSONDecodeError as e:
              logger.warning(f"JSON解析失败: {e}，返回原始文本")
              return {
                  "analysis": "解析失败",
                  "guidance": content,
                  "action_items": [],
                  "needs_another_iteration": False,
                  "resolved": False,
                  "requires_human_intervention": True,
              }
  
      except Exception as e:
          logger.error(f"API请求失败: {e}")
          return {
              "analysis": f"请求失败: {str(e)}",
              "guidance": "请检查API密钥、Base URL和网络连接",
              "action_items": [],
              "needs_another_iteration": False,
              "resolved": False,
              "requires_human_intervention": True,
          }