MCPcopy Create free account
hub / github.com/ScaleML/AgentSPEX / get_step_instruction

Function get_step_instruction

src/harness/utils/step_utils.py:14–66  ·  view source on GitHub ↗
(step_data: Dict[str, Any], context: Dict[str, Any])

Source from the content-addressed store, hash-verified

12
13
14def get_step_instruction(step_data: Dict[str, Any], context: Dict[str, Any]) -> str:
15 if not isinstance(step_data, dict):
16 return ""
17
18 step_type = step_data.get("step_type")
19 if not step_type:
20 return ""
21
22 def expand(value: str) -> str:
23 return expand_template_variables(value or "", context)
24
25 simple_instruction_types = {
26 "step",
27 "task",
28 }
29
30 if step_type in simple_instruction_types:
31 return expand((step_data.get(step_type) or {}).get("instruction"))
32
33 if step_type == "for_each":
34 fe = step_data.get("for_each") or {}
35 return f"{fe.get('variable', '')} in {fe.get('in', '')}".strip()
36
37 if step_type in {"if", "while"}:
38 return f"{step_type} {(step_data.get(step_type) or {}).get('condition', '')}".strip()
39
40 if step_type == "switch":
41 return f"switch {(step_data.get('switch') or {}).get('variable', '')}".strip()
42
43 if step_type in {"call", "gather"}:
44 return expand((step_data.get(step_type) or {}).get("module"))
45
46 if step_type == "input":
47 cfg, _ = split_step_value(step_data, "input")
48 return expand(cfg.get("prompt"))
49
50 if step_type == "increment":
51 cfg, scalar = split_step_value(step_data, step_type)
52 variable = cfg.get("variable") or (scalar if isinstance(scalar, str) else "")
53 return f"{step_type} {variable}".strip()
54
55 if step_type == "set_variable":
56 cfg, scalar = split_step_value(step_data, "set_variable")
57 if scalar is not None:
58 return f"set {scalar}".strip()
59 variable = cfg.get("variable") or cfg.get("name", "")
60 value = cfg.get("value", "")
61 return f"set {variable} = {value}".strip()
62
63 if step_type == "return":
64 return _render_return(step_data, context)
65
66 return ""
67
68
69def _render_return(step_data: Dict[str, Any], context: Dict[str, Any]) -> str:

Callers 1

execute_workflow_stepMethod · 0.85

Calls 4

expandFunction · 0.85
split_step_valueFunction · 0.85
_render_returnFunction · 0.85
getMethod · 0.80

Tested by

no test coverage detected