| class ExecutorContext: | |
| """Holds state for the execution engine, e.g. mapping step IDs to their outputs.""" | |
| def __init__(self): | |
| self.step_outputs = {} | |
| def set_output(self, step_id: str, output: str): | |
| self.step_outputs[step_id] = output | |
| def get_output(self, step_id: str) -> str: | |
| return self.step_outputs.get(step_id, "") | |
| def resolve_placeholders(self, text: str) -> str: | |
| """ | |
| Replaces {{step_id.output}} with the actual output of step_id. | |
| """ | |
| if not isinstance(text, str): | |
| return text | |
| for s_id, out in self.step_outputs.items(): | |
| placeholder = f"{{{{{s_id}.output}}}}" | |
| if placeholder in text: | |
| text = text.replace(placeholder, str(out)) | |
| return text | |
| def resolve_dict(self, parameters: dict) -> dict: | |
| """Resolves placeholders recursively in a parameters dict for tools.""" | |
| resolved = {} | |
| for k, v in parameters.items(): | |
| if isinstance(v, str): | |
| resolved[k] = self.resolve_placeholders(v) | |
| else: | |
| resolved[k] = v | |
| return resolved | |