Single agent with basic tools and MCP support.
| 43 | |
| 44 | |
| 45 | class Agent: |
| 46 | """Single agent with basic tools and MCP support.""" |
| 47 | |
| 48 | def __init__( |
| 49 | self, |
| 50 | llm_client: LLMClient, |
| 51 | system_prompt: str, |
| 52 | tools: list[Tool], |
| 53 | max_steps: int = 50, |
| 54 | workspace_dir: str = "./workspace", |
| 55 | token_limit: int = 80000, # Summary triggered when tokens exceed this value |
| 56 | ): |
| 57 | self.llm = llm_client |
| 58 | self.tools = {tool.name: tool for tool in tools} |
| 59 | self.max_steps = max_steps |
| 60 | self.token_limit = token_limit |
| 61 | self.workspace_dir = Path(workspace_dir) |
| 62 | # Cancellation event for interrupting agent execution (set externally, e.g., by Esc key) |
| 63 | self.cancel_event: Optional[asyncio.Event] = None |
| 64 | |
| 65 | # Ensure workspace exists |
| 66 | self.workspace_dir.mkdir(parents=True, exist_ok=True) |
| 67 | |
| 68 | # Inject workspace information into system prompt if not already present |
| 69 | if "Current Workspace" not in system_prompt: |
| 70 | workspace_info = f"\n\n## Current Workspace\nYou are currently working in: `{self.workspace_dir.absolute()}`\nAll relative paths will be resolved relative to this directory." |
| 71 | system_prompt = system_prompt + workspace_info |
| 72 | |
| 73 | self.system_prompt = system_prompt |
| 74 | |
| 75 | # Initialize message history |
| 76 | self.messages: list[Message] = [Message(role="system", content=system_prompt)] |
| 77 | |
| 78 | # Initialize logger |
| 79 | self.logger = AgentLogger() |
| 80 | |
| 81 | # Token usage from last API response (updated after each LLM call) |
| 82 | self.api_total_tokens: int = 0 |
| 83 | # Flag to skip token check right after summary (avoid consecutive triggers) |
| 84 | self._skip_next_token_check: bool = False |
| 85 | |
| 86 | def add_user_message(self, content: str): |
| 87 | """Add a user message to history.""" |
| 88 | self.messages.append(Message(role="user", content=content)) |
| 89 | |
| 90 | def _check_cancelled(self) -> bool: |
| 91 | """Check if agent execution has been cancelled. |
| 92 | |
| 93 | Returns: |
| 94 | True if cancelled, False otherwise. |
| 95 | """ |
| 96 | if self.cancel_event is not None and self.cancel_event.is_set(): |
| 97 | return True |
| 98 | return False |
| 99 | |
| 100 | def _cleanup_incomplete_messages(self): |
| 101 | """Remove the incomplete assistant message and its partial tool results. |
| 102 |
no outgoing calls