Process a user message through the agent loop, returning the final text response.
(self, user_message: str)
| 58 | self.context.set_system_prompt(system_prompt) |
| 59 | |
| 60 | def run(self, user_message: str) -> str: |
| 61 | """Process a user message through the agent loop, returning the final text response.""" |
| 62 | self.context.add_user_message(user_message) |
| 63 | final_text = "" |
| 64 | |
| 65 | for turn in range(self.config.max_turns): |
| 66 | response = self._call_api() |
| 67 | tool_calls, text_parts = self._parse_response(response) |
| 68 | |
| 69 | if text_parts: |
| 70 | final_text = "\n".join(text_parts) |
| 71 | |
| 72 | if not tool_calls: |
| 73 | # No tool calls -- the loop ends, return the text |
| 74 | self.context.add_assistant_message(response.content) |
| 75 | break |
| 76 | |
| 77 | # There are tool calls -- execute them and continue the loop |
| 78 | self.context.add_assistant_message(response.content) |
| 79 | self._execute_tool_calls(tool_calls) |
| 80 | else: |
| 81 | if not final_text: |
| 82 | final_text = "(max turns reached without a final response)" |
| 83 | |
| 84 | return final_text |
| 85 | |
| 86 | def _call_api(self) -> Any: |
| 87 | """Call the Anthropic API with current context.""" |
no test coverage detected