Function _extract_user_content

src/google/adk/cli/agent_test_runner.py:280–329

Extracts user content from an event dict and returns a types.Content object. Agent-emitted user-role events (e.g., task FRs synthesized by the Task Delegation API wrapper) are skipped — those are produced by the agent itself and carry a non-empty ``nodeInfo.path``. Re-feeding them to ``run

(event: dict)

def _extract_user_content(event: dict) -> Optional[types.Content]:
  """Extracts user content from an event dict and returns a types.Content object.

  Agent-emitted user-role events (e.g., task FRs synthesized by the Task
  Delegation API wrapper) are skipped — those are produced by the agent
  itself and carry a non-empty ``nodeInfo.path``.  Re-feeding them to
  ``runner.run`` would trigger extra LLM calls.
  """
  if event.get("author") != "user":
    return None

  # Real external user input has no node path; agent-emitted events do.
  if event.get("nodeInfo", {}).get("path"):
    return None

  content_dict = event.get("content", {})
  if not content_dict:
    return None

  parts = content_dict.get("parts", [])
  real_parts = []
  for p in parts:
    if "functionResponse" in p:
      fr = p["functionResponse"]
      real_parts.append(
          types.Part(
              function_response=types.FunctionResponse(
                  id=fr.get("id"),
                  name=fr.get("name"),
                  response=fr.get("response"),
              )
          )
      )
    elif "text" in p:
      real_parts.append(types.Part(text=p["text"]))
    elif "functionCall" in p:
      fc = p["functionCall"]
      real_parts.append(
          types.Part(
              function_call=types.FunctionCall(
                  id=fc.get("id"),
                  name=fc.get("name"),
                  args=fc.get("args"),
              )
          )
      )

  if real_parts:
    return types.Content(role="user", parts=real_parts)
  return None


def _remap_node_path(path: str, id_map: dict[str, str]) -> str: