Minimal ACP adapter wrapping the existing Agent runtime.
| 68 | |
| 69 | |
| 70 | class MiniMaxACPAgent: |
| 71 | """Minimal ACP adapter wrapping the existing Agent runtime.""" |
| 72 | |
| 73 | def __init__( |
| 74 | self, |
| 75 | conn: AgentSideConnection, |
| 76 | config: Config, |
| 77 | llm: LLMClient, |
| 78 | base_tools: list, |
| 79 | system_prompt: str, |
| 80 | ): |
| 81 | self._conn = conn |
| 82 | self._config = config |
| 83 | self._llm = llm |
| 84 | self._base_tools = base_tools |
| 85 | self._system_prompt = system_prompt |
| 86 | self._sessions: dict[str, SessionState] = {} |
| 87 | |
| 88 | async def initialize(self, params: InitializeRequest) -> InitializeResponse: # noqa: ARG002 |
| 89 | return InitializeResponse( |
| 90 | protocolVersion=PROTOCOL_VERSION, |
| 91 | agentCapabilities=AgentCapabilities(loadSession=False), |
| 92 | agentInfo=Implementation(name="mini-agent", title="Mini-Agent", version="0.1.0"), |
| 93 | ) |
| 94 | |
| 95 | async def newSession(self, params: NewSessionRequest) -> NewSessionResponse: |
| 96 | session_id = f"sess-{len(self._sessions)}-{uuid4().hex[:8]}" |
| 97 | workspace = Path(params.cwd or self._config.agent.workspace_dir).expanduser() |
| 98 | if not workspace.is_absolute(): |
| 99 | workspace = workspace.resolve() |
| 100 | tools = list(self._base_tools) |
| 101 | add_workspace_tools(tools, self._config, workspace) |
| 102 | agent = Agent(llm_client=self._llm, system_prompt=self._system_prompt, tools=tools, max_steps=self._config.agent.max_steps, workspace_dir=str(workspace)) |
| 103 | self._sessions[session_id] = SessionState(agent=agent) |
| 104 | return NewSessionResponse(sessionId=session_id) |
| 105 | |
| 106 | async def prompt(self, params: PromptRequest) -> PromptResponse: |
| 107 | state = self._sessions.get(params.sessionId) |
| 108 | if not state: |
| 109 | # Auto-create session if not found (compatibility with clients that skip newSession) |
| 110 | logger.warning(f"Session '{params.sessionId}' not found, auto-creating new session") |
| 111 | new_session = await self.newSession(NewSessionRequest(cwd=None)) |
| 112 | state = self._sessions.get(new_session.sessionId) |
| 113 | if not state: |
| 114 | logger.error("Failed to auto-create session") |
| 115 | return PromptResponse(stopReason="refusal") |
| 116 | state.cancelled = False |
| 117 | user_text = "\n".join(block.get("text", "") if isinstance(block, dict) else getattr(block, "text", "") for block in params.prompt) |
| 118 | state.agent.messages.append(Message(role="user", content=user_text)) |
| 119 | stop_reason = await self._run_turn(state, params.sessionId) |
| 120 | return PromptResponse(stopReason=stop_reason) |
| 121 | |
| 122 | async def cancel(self, params: CancelNotification) -> None: |
| 123 | state = self._sessions.get(params.sessionId) |
| 124 | if state: |
| 125 | state.cancelled = True |
| 126 | |
| 127 | async def _run_turn(self, state: SessionState, session_id: str) -> str: |
no outgoing calls