Compare commits

...

11 Commits

Author SHA1 Message Date
Rohit Malhotra
953f99a147 CLI(V1): resume conversations (#11154)
Co-authored-by: openhands <openhands@all-hands.dev>
2025-10-02 04:11:12 +08:00
Xingyao Wang
1d78513407 v1 CLI: fix anthropic thinking issue (#11207) 2025-10-02 01:12:54 +08:00
Xingyao Wang
d51c6bb992 Update OpenHands CLI for agent SDK refactor (#11165) 2025-10-01 23:15:47 +08:00
Xingyao Wang
1cd8eada2b Fix: Allow Ctrl+C to cancel settings configuration prompts (#11201)
Co-authored-by: openhands <openhands@all-hands.dev>
2025-10-01 09:40:12 -04:00
Xingyao Wang
44c4e0e5fd Revert "Update OpenHands CLI for agent SDK refactor"
This reverts commit a9982f96c6.
2025-09-28 17:02:18 -04:00
openhands
a9982f96c6 Update OpenHands CLI for agent SDK refactor
- Update preset imports from openhands.sdk.preset to openhands.tools.preset
- Bump agent SDK and tools dependencies to latest commit (004f381a)
- Add LLM metadata integration with get_llm_metadata utility function
- Pass correct metadata to LLM initialization including agent name, session ID, and version info
- Update AgentStore to refresh LLM metadata on agent load

Co-authored-by: openhands <openhands@all-hands.dev>
2025-09-28 17:41:42 +00:00
Rohit Malhotra
7112b4e329 CLI(V1): Multiline inputs (#11131)
Co-authored-by: openhands <openhands@all-hands.dev>
2025-09-26 13:33:21 -04:00
Rohit Malhotra
c2d1d15a8f CLI(V1): Add loading screen + suppress extraneous logs (#11134)
Co-authored-by: openhands <openhands@all-hands.dev>
2025-09-26 11:05:52 -04:00
Rohit Malhotra
d2bb882c96 Add /mcp command for MCP server configuration in CLI (#11105)
Co-authored-by: openhands <openhands@all-hands.dev>
2025-09-25 16:58:33 -04:00
Rohit Malhotra
e995882194 CLI(V1): session persistence (#11129)
Co-authored-by: openhands <openhands@all-hands.dev>
2025-09-25 16:26:13 -04:00
Rohit Malhotra
ef1441bbe5 CLI(V1): restore terminal state (#11127)
Co-authored-by: Xingyao Wang <xingyao@all-hands.dev>
2025-09-25 13:48:57 -04:00
28 changed files with 1017 additions and 161 deletions

View File

@@ -13,13 +13,18 @@ import subprocess
import sys
from pathlib import Path
from openhands_cli.locations import PERSISTENCE_DIR, WORK_DIR, AGENT_SETTINGS_PATH
from openhands.sdk.preset.default import get_default_agent
from openhands_cli.llm_utils import get_llm_metadata
from openhands.tools.preset.default import get_default_agent
from openhands.sdk import LLM
import time
import select
dummy_agent = get_default_agent(
llm=LLM(model='dummy-model', api_key='dummy-key'),
llm=LLM(
model='dummy-model',
api_key='dummy-key',
metadata=get_llm_metadata(model_name='dummy-model', agent_name='openhands-cli')
),
working_dir=WORK_DIR,
persistence_dir=PERSISTENCE_DIR,
cli_mode=True

View File

@@ -4,35 +4,54 @@ Agent chat functionality for OpenHands CLI.
Provides a conversation interface with an AI agent using OpenHands patterns.
"""
import logging
import uuid
import sys
from openhands.sdk import (
Message,
TextContent,
AgentContext,
)
from openhands.sdk.conversation.state import AgentExecutionStatus
from prompt_toolkit import PromptSession, print_formatted_text
from openhands_cli.tui.settings.mcp_screen import MCPScreen
from openhands_cli.user_actions.utils import get_session_prompter
from prompt_toolkit import print_formatted_text
from prompt_toolkit.formatted_text import HTML
from openhands_cli.runner import ConversationRunner
from openhands_cli.setup import setup_conversation, MissingAgentSpec
from openhands_cli.tui.settings.settings_screen import SettingsScreen
from openhands_cli.tui.tui import (
CommandCompleter,
display_help,
display_welcome,
)
from openhands_cli.user_actions import UserConfirmation, exit_session_confirmation
from openhands_cli.locations import WORK_DIR
logger = logging.getLogger(__name__)
def run_cli_entry() -> None:
def _restore_tty() -> None:
"""
Ensure terminal modes are reset in case prompt_toolkit cleanup didn't run.
- Turn off application cursor keys (DECCKM): ESC[?1l
- Turn off bracketed paste: ESC[?2004l
"""
try:
sys.stdout.write("\x1b[?1l\x1b[?2004l")
sys.stdout.flush()
except Exception:
pass
def _print_exit_hint(conversation_id: str) -> None:
"""Print a resume hint with the current conversation ID."""
print_formatted_text(HTML(f"<grey>Conversation ID:</grey> <yellow>{conversation_id}</yellow>"))
print_formatted_text(
HTML(
f"<grey>Hint:</grey> run <gold>openhands-cli --resume {conversation_id}</gold> "
"to resume this conversation."
)
)
def run_cli_entry(resume_conversation_id: str | None = None) -> None:
"""Run the agent chat session using the agent SDK.
Raises:
AgentSetupError: If agent setup fails
KeyboardInterrupt: If user interrupts the session
@@ -44,20 +63,15 @@ def run_cli_entry() -> None:
while not conversation:
try:
conversation = setup_conversation()
conversation = setup_conversation(resume_conversation_id)
except MissingAgentSpec:
settings_screen.handle_basic_settings(escapable=False)
# Generate session ID
session_id = str(uuid.uuid4())[:8]
display_welcome(session_id)
# Create prompt session with command completer
session = PromptSession(completer=CommandCompleter())
display_welcome(conversation.id, bool(resume_conversation_id))
# Create conversation runner to handle state machine logic
runner = ConversationRunner(conversation)
session = get_session_prompter()
# Main chat loop
while True:
@@ -83,6 +97,7 @@ def run_cli_entry() -> None:
exit_confirmation = exit_session_confirmation()
if exit_confirmation == UserConfirmation.ACCEPT:
print_formatted_text(HTML("\n<yellow>Goodbye! 👋</yellow>"))
_print_exit_hint(conversation.id)
break
elif command == "/settings":
@@ -90,14 +105,21 @@ def run_cli_entry() -> None:
settings_screen.display_settings()
continue
elif command == "/clear":
display_welcome(session_id)
elif command == "/mcp":
mcp_screen = MCPScreen()
mcp_screen.display_mcp_info(conversation.agent)
continue
elif command == "/clear":
display_welcome(conversation.id)
continue
elif command == "/help":
display_help()
continue
elif command == "/status":
print_formatted_text(HTML(f"<grey>Session ID: {session_id}</grey>"))
print_formatted_text(HTML(f"<grey>Conversation ID: {conversation.id}</grey>"))
print_formatted_text(HTML("<grey>Status: Active</grey>"))
confirmation_status = (
"enabled" if conversation.state.confirmation_mode else "disabled"
@@ -106,6 +128,7 @@ def run_cli_entry() -> None:
HTML(f"<grey>Confirmation mode: {confirmation_status}</grey>")
)
continue
elif command == "/confirm":
runner.toggle_confirmation_mode()
new_status = "enabled" if runner.is_confirmation_mode_enabled else "disabled"
@@ -113,13 +136,7 @@ def run_cli_entry() -> None:
HTML(f"<yellow>Confirmation mode {new_status}</yellow>")
)
continue
elif command == "/new":
print_formatted_text(
HTML("<yellow>Starting new conversation...</yellow>")
)
session_id = str(uuid.uuid4())[:8]
display_welcome(session_id)
continue
elif command == "/resume":
if not (
conversation.state.agent_status == AgentExecutionStatus.PAUSED
@@ -129,7 +146,6 @@ def run_cli_entry() -> None:
print_formatted_text(
HTML("<red>No paused conversation to resume...</red>")
)
continue
# Resume without new message
@@ -143,4 +159,10 @@ def run_cli_entry() -> None:
exit_confirmation = exit_session_confirmation()
if exit_confirmation == UserConfirmation.ACCEPT:
print_formatted_text(HTML("\n<yellow>Goodbye! 👋</yellow>"))
_print_exit_hint(conversation.id)
break
# Clean up terminal state
_restore_tty()

View File

@@ -1,5 +1,7 @@
from openhands_cli.listeners.pause_listener import PauseListener
from openhands_cli.listeners.loading_listener import LoadingContext
__all__ = [
"PauseListener",
"LoadingContext"
]

View File

@@ -0,0 +1,61 @@
"""
Loading animation utilities for OpenHands CLI.
Provides animated loading screens during agent initialization.
"""
import sys
import threading
import time
def display_initialization_animation(text: str, is_loaded: threading.Event) -> None:
"""Display a spinning animation while agent is being initialized.
Args:
text: The text to display alongside the animation
is_loaded: Threading event that signals when loading is complete
"""
ANIMATION_FRAMES = ['', '', '', '', '', '', '', '', '', '']
i = 0
while not is_loaded.is_set():
sys.stdout.write('\n')
sys.stdout.write(
f'\033[s\033[J\033[38;2;255;215;0m[{ANIMATION_FRAMES[i % len(ANIMATION_FRAMES)]}] {text}\033[0m\033[u\033[1A'
)
sys.stdout.flush()
time.sleep(0.1)
i += 1
sys.stdout.write('\r' + ' ' * (len(text) + 10) + '\r')
sys.stdout.flush()
class LoadingContext:
"""Context manager for displaying loading animations in a separate thread."""
def __init__(self, text: str):
"""Initialize the loading context.
Args:
text: The text to display during loading
"""
self.text = text
self.is_loaded = threading.Event()
self.loading_thread: threading.Thread | None = None
def __enter__(self) -> 'LoadingContext':
"""Start the loading animation in a separate thread."""
self.loading_thread = threading.Thread(
target=display_initialization_animation,
args=(self.text, self.is_loaded),
daemon=True
)
self.loading_thread.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Stop the loading animation and clean up the thread."""
self.is_loaded.set()
if self.loading_thread:
self.loading_thread.join(timeout=1.0) # Wait up to 1 second for thread to finish

View File

@@ -2,7 +2,7 @@ import threading
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from openhands.sdk import Conversation
from openhands.sdk import BaseConversation
from prompt_toolkit import HTML, print_formatted_text
from prompt_toolkit.input import Input, create_input
from prompt_toolkit.keys import Keys
@@ -71,7 +71,7 @@ class PauseListener(threading.Thread):
@contextmanager
def pause_listener(
conversation: Conversation, input_source: Input | None = None
conversation: BaseConversation, input_source: Input | None = None
) -> Iterator[PauseListener]:
"""Ensure PauseListener always starts/stops cleanly."""
listener = PauseListener(on_pause=conversation.pause, input_source=input_source)

View File

@@ -0,0 +1,55 @@
"""Utility functions for LLM configuration in OpenHands CLI."""
import os
from typing import Any
def get_llm_metadata(
model_name: str,
llm_type: str,
session_id: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
"""
Generate LLM metadata for OpenHands CLI.
Args:
model_name: Name of the LLM model
agent_name: Name of the agent (defaults to "openhands-cli")
session_id: Optional session identifier
user_id: Optional user identifier
Returns:
Dictionary containing metadata for LLM initialization
"""
# Import here to avoid circular imports
openhands_sdk_version: str = "n/a"
try:
import openhands.sdk
openhands_sdk_version = openhands.sdk.__version__
except (ModuleNotFoundError, AttributeError):
pass
openhands_tools_version: str = "n/a"
try:
import openhands.tools
openhands_tools_version = openhands.tools.__version__
except (ModuleNotFoundError, AttributeError):
pass
metadata = {
"trace_version": openhands_sdk_version,
"tags": [
"app:openhands-cli",
f"model:{model_name}",
f"type:{llm_type}",
f"web_host:{os.environ.get('WEB_HOST', 'unspecified')}",
f"openhands_sdk_version:{openhands_sdk_version}",
f"openhands_tools_version:{openhands_tools_version}",
],
}
if session_id is not None:
metadata["session_id"] = session_id
if user_id is not None:
metadata["trace_user_id"] = user_id
return metadata

View File

@@ -1,9 +1,14 @@
import os
from uuid import UUID
# Configuration directory for storing agent settings and CLI configuration
PERSISTENCE_DIR = os.path.expanduser("~/.openhands")
CONVERSATIONS_DIR = os.path.join(PERSISTENCE_DIR, "conversations")
# Working directory for agent operations (current directory where CLI is run)
WORK_DIR = os.getcwd()
AGENT_SETTINGS_PATH = "agent_settings.json"
# MCP configuration file (relative to PERSISTENCE_DIR)
MCP_CONFIG_FILE = "mcp.json"

View File

@@ -24,6 +24,7 @@ def get_cli_style() -> BaseStyle:
"completion-menu.completion.current fuzzymatch.outside": "fg:#ffffff bg:#888888",
"selected": COLOR_GOLD,
"risk-high": "#FF0000 bold", # Red bold for HIGH risk
"placeholder": "#888888 italic",
}
)
return merge_styles([base, custom])

View File

@@ -5,8 +5,7 @@ from openhands.sdk.security.confirmation_policy import (
ConfirmRisky,
ConfirmationPolicyBase
)
from openhands.sdk.conversation.state import AgentExecutionStatus
from openhands.sdk.event.utils import get_unmatched_actions
from openhands.sdk.conversation.state import AgentExecutionStatus, ConversationState
from prompt_toolkit import HTML, print_formatted_text
from openhands_cli.listeners.pause_listener import PauseListener, pause_listener
@@ -117,7 +116,7 @@ class ConversationRunner:
UserConfirmation indicating the user's choice
"""
pending_actions = get_unmatched_actions(self.conversation.state.events)
pending_actions = ConversationState.get_unmatched_actions(self.conversation.state.events)
if not pending_actions:
return UserConfirmation.ACCEPT

View File

@@ -1,13 +1,14 @@
from openhands.sdk import (
Conversation,
BaseConversation
)
from openhands_cli.tui.settings.store import AgentStore
from prompt_toolkit import HTML, print_formatted_text
import uuid
from openhands.sdk import BaseConversation, Conversation, Workspace, register_tool
from openhands.tools.execute_bash import BashTool
from openhands.tools.str_replace_editor import FileEditorTool
from openhands.tools.task_tracker import TaskTrackerTool
from openhands.sdk import register_tool
from prompt_toolkit import HTML, print_formatted_text
from openhands_cli.listeners import LoadingContext
from openhands_cli.locations import CONVERSATIONS_DIR, WORK_DIR
from openhands_cli.tui.settings.store import AgentStore
register_tool("BashTool", BashTool)
register_tool("FileEditorTool", FileEditorTool)
@@ -18,21 +19,43 @@ class MissingAgentSpec(Exception):
"""Raised when agent specification is not found or invalid."""
pass
def setup_conversation() -> BaseConversation:
def setup_conversation(conversation_id: str | None = None) -> BaseConversation:
"""
Setup the conversation with agent.
Args:
conversation_id: conversation ID to use. If not provided, a random UUID will be generated.
Raises:
MissingAgentSpec: If agent specification is not found or invalid.
"""
agent_store = AgentStore()
agent = agent_store.load()
if not agent:
raise MissingAgentSpec("Agent specification not found. Please configure your agent settings.")
# Use provided conversation_id or generate a random one
if conversation_id is None:
conversation_id = uuid.uuid4()
elif isinstance(conversation_id, str):
try:
conversation_id = uuid.UUID(conversation_id)
except ValueError as e:
print_formatted_text(
HTML(f"<yellow>Warning: '{conversation_id}' is not a valid UUID.</yellow>")
)
raise e
# Create conversation - agent context is now set in AgentStore.load()
conversation = Conversation(agent=agent)
with LoadingContext("Initializing OpenHands agent..."):
agent_store = AgentStore()
agent = agent_store.load(session_id=str(conversation_id))
if not agent:
raise MissingAgentSpec("Agent specification not found. Please configure your agent settings.")
# Create conversation - agent context is now set in AgentStore.load()
conversation = Conversation(
agent=agent,
workspace=Workspace(working_dir=WORK_DIR),
# Conversation will add /<conversation_id> to this path
persistence_dir=CONVERSATIONS_DIR,
conversation_id=conversation_id
)
print_formatted_text(
HTML(f"<green>✓ Agent initialized with model: {agent.llm.model}</green>")

View File

@@ -4,6 +4,14 @@ Simple main entry point for OpenHands CLI.
This is a simplified version that demonstrates the TUI functionality.
"""
import argparse
import logging
import os
debug_env = os.getenv('DEBUG', 'false').lower()
if debug_env != '1' and debug_env != 'true':
logging.disable(logging.WARNING)
from prompt_toolkit import print_formatted_text
from prompt_toolkit.formatted_text import HTML
from openhands_cli.agent_chat import run_cli_entry
@@ -16,9 +24,20 @@ def main() -> None:
ImportError: If agent chat dependencies are missing
Exception: On other error conditions
"""
parser = argparse.ArgumentParser(
description="OpenHands CLI - Terminal User Interface for OpenHands AI Agent"
)
parser.add_argument(
"--resume",
type=str,
help="Conversation ID to use for the session. If not provided, a random UUID will be generated."
)
args = parser.parse_args()
try:
# Start agent chat directly by default
run_cli_entry()
# Start agent chat
run_cli_entry(resume_conversation_id=args.resume)
except ImportError as e:
print_formatted_text(
@@ -35,6 +54,7 @@ def main() -> None:
except Exception as e:
print_formatted_text(HTML(f"<red>Error starting agent chat: {e}</red>"))
import traceback
traceback.print_exc()
raise

View File

@@ -0,0 +1,202 @@
import json
from pathlib import Path
from typing import Any
from openhands_cli.locations import MCP_CONFIG_FILE, PERSISTENCE_DIR
from openhands_cli.tui.settings.store import AgentStore
from prompt_toolkit import HTML, print_formatted_text
from fastmcp.mcp_config import MCPConfig
from openhands.sdk import Agent
class MCPScreen:
"""
MCP Screen
1. Display information about setting up MCP
2. See existing servers that are setup
3. Debug additional servers passed via mcp.json
4. Identify servers waiting to sync on session restart
"""
# ---------- server spec handlers ----------
def _check_server_specs_are_equal(
self,
first_server_spec,
second_server_spec
) -> bool:
first_stringified_server_spec = json.dumps(first_server_spec, sort_keys=True)
second_stringified_server_spec = json.dumps(second_server_spec, sort_keys=True)
return first_stringified_server_spec == second_stringified_server_spec
def _check_mcp_config_status(self) -> dict:
"""Check the status of the MCP configuration file and return information about it."""
config_path = Path(PERSISTENCE_DIR) / MCP_CONFIG_FILE
if not config_path.exists():
return {
"exists": False,
"valid": False,
"servers": {},
"message": f"MCP configuration file not found at ~/.openhands/{MCP_CONFIG_FILE}",
}
try:
mcp_config = MCPConfig.from_file(config_path)
servers = mcp_config.to_dict().get("mcpServers", {})
return {
"exists": True,
"valid": True,
"servers": servers,
"message": f"Valid MCP configuration found with {len(servers)} server(s)",
}
except Exception as e:
return {
"exists": True,
"valid": False,
"servers": {},
"message": f"Invalid MCP configuration file: {str(e)}",
}
# ---------- TUI helpers ----------
def _get_mcp_server_diff(
self,
current: dict[str, Any],
incoming: dict[str, Any],
) -> None:
"""
Display a diff-style view:
- Always show the MCP servers the agent is *currently* configured with
- If there are incoming servers (from ~/.openhands/mcp.json),
clearly show which ones are NEW (not in current) and which ones are CHANGED
(same name but different config). Unchanged servers are not repeated.
"""
print_formatted_text(HTML("<white>Current Agent MCP Servers:</white>"))
if current:
for name, cfg in current.items():
self._render_server_summary(name, cfg, indent=2)
else:
print_formatted_text(HTML(" <yellow>None configured on the current agent.</yellow>"))
print_formatted_text("")
# If no incoming, we're done
if not incoming:
print_formatted_text(HTML("<grey>No incoming servers detected for next restart.</grey>"))
print_formatted_text("")
return
# Compare names and configs
current_names = set(current.keys())
incoming_names = set(incoming.keys())
new_servers = sorted(incoming_names - current_names)
overriden_servers = []
for name in sorted(incoming_names & current_names):
if not self._check_server_specs_are_equal(current[name], incoming[name]):
overriden_servers.append(name)
# Display incoming section header
print_formatted_text(HTML("<white>Incoming Servers on Restart (from ~/.openhands/mcp.json):</white>"))
if not new_servers and not overriden_servers:
print_formatted_text(HTML(" <grey>All configured servers match the current agent configuration.</grey>"))
print_formatted_text("")
return
if new_servers:
print_formatted_text(HTML(" <green>New servers (will be added):</green>"))
for name in new_servers:
self._render_server_summary(name, incoming[name], indent=4)
if overriden_servers:
print_formatted_text(HTML(" <yellow>Updated servers (configuration will change):</yellow>"))
for name in overriden_servers:
print_formatted_text(HTML(f" <white>• {name}</white>"))
print_formatted_text(HTML(" <grey>Current:</grey>"))
self._render_server_summary(None, current[name], indent=8)
print_formatted_text(HTML(" <grey>Incoming:</grey>"))
self._render_server_summary(None, incoming[name], indent=8)
print_formatted_text("")
def _render_server_summary(
self,
server_name: str | None,
server_spec: dict[str, Any],
indent: int = 2
) -> None:
pad = " " * indent
if server_name:
print_formatted_text(HTML(f"{pad}<white>• {server_name}</white>"))
if isinstance(server_spec, dict):
if "command" in server_spec:
cmd = server_spec.get("command", "")
args = server_spec.get("args", [])
args_str = " ".join(args) if args else ""
print_formatted_text(HTML(f"{pad} <grey>Type: Command-based</grey>"))
if cmd or args_str:
print_formatted_text(HTML(f"{pad} <grey>Command: {cmd} {args_str}</grey>"))
elif "url" in server_spec:
url = server_spec.get("url", "")
auth = server_spec.get("auth", "none")
print_formatted_text(HTML(f"{pad} <grey>Type: URL-based</grey>"))
if url:
print_formatted_text(HTML(f"{pad} <grey>URL: {url}</grey>"))
print_formatted_text(HTML(f"{pad} <grey>Auth: {auth}</grey>"))
def _display_information_header(self) -> None:
print_formatted_text(HTML("<gold>MCP (Model Context Protocol) Configuration</gold>"))
print_formatted_text("")
print_formatted_text(HTML("<white>To get started:</white>"))
print_formatted_text(HTML(" 1. Create the configuration file: <cyan>~/.openhands/mcp.json</cyan>"))
print_formatted_text(
HTML(
" 2. Add your MCP server configurations "
"<cyan>https://gofastmcp.com/clients/client#configuration-format</cyan>"
)
)
print_formatted_text(HTML(" 3. Restart your OpenHands session to load the new configuration"))
print_formatted_text("")
# ---------- status + display entrypoint ----------
def display_mcp_info(self, existing_agent: Agent) -> None:
"""Display comprehensive MCP configuration information."""
self._display_information_header()
# Always determine current & incoming first
status = self._check_mcp_config_status()
incoming_servers = status.get("servers", {}) if status.get("valid") else {}
current_servers = existing_agent.mcp_config.get('mcpServers', {})
# Show file status
if not status["exists"]:
print_formatted_text(HTML("<yellow>Status: Configuration file not found</yellow>"))
elif not status["valid"]:
print_formatted_text(HTML(f"<red>Status: {status['message']}</red>"))
print_formatted_text("")
print_formatted_text(HTML("<white>Please check your configuration file format.</white>"))
else:
print_formatted_text(HTML(f"<green>Status: {status['message']}</green>"))
print_formatted_text("")
# Always show the agent's current servers
# Then show incoming (deduped and changes highlighted)
self._get_mcp_server_diff(current_servers, incoming_servers)

View File

@@ -1,5 +1,6 @@
import os
from openhands_cli.locations import AGENT_SETTINGS_PATH, PERSISTENCE_DIR, WORK_DIR
from openhands_cli.llm_utils import get_llm_metadata
from openhands_cli.tui.settings.store import AgentStore
from openhands_cli.user_actions.settings_action import (
SettingsType,
@@ -14,15 +15,16 @@ from openhands_cli.user_actions.settings_action import (
)
from openhands_cli.tui.utils import StepCounter
from prompt_toolkit import HTML, print_formatted_text
from openhands.sdk import Conversation, LLM, LocalFileStore
from openhands.sdk.preset.default import get_default_agent
from openhands.sdk import BaseConversation, LLM, LocalFileStore
from openhands.sdk.security.confirmation_policy import NeverConfirm
from openhands.tools.preset.default import get_default_agent
from prompt_toolkit.shortcuts import print_container
from prompt_toolkit.widgets import Frame, TextArea
from openhands_cli.pt_style import COLOR_GREY
class SettingsScreen:
def __init__(self, conversation: Conversation | None = None):
def __init__(self, conversation: BaseConversation | None = None):
self.file_store = LocalFileStore(PERSISTENCE_DIR)
self.agent_store = AgentStore()
self.conversation = conversation
@@ -31,6 +33,8 @@ class SettingsScreen:
agent_spec = self.agent_store.load()
if not agent_spec:
return
assert self.conversation is not None, \
"Conversation must be set to display settings."
llm = agent_spec.llm
advanced_llm_settings = True if llm.base_url else False
@@ -57,7 +61,7 @@ class SettingsScreen:
)
labels_and_values.extend([
(" API Key", "********" if llm.api_key else "Not Set"),
(" Confirmation Mode", "Enabled" if self.conversation.confirmation_policy_active else "Disabled"),
(" Confirmation Mode", "Enabled" if not isinstance(self.conversation.state.confirmation_policy, NeverConfirm) else "Disabled"),
(" Memory Condensation", "Enabled" if agent_spec.condenser else "Disabled"),
(" Configuration File", os.path.join(PERSISTENCE_DIR, AGENT_SETTINGS_PATH))
])
@@ -114,7 +118,7 @@ class SettingsScreen:
api_key = prompt_api_key(
step_counter,
provider,
self.conversation.agent.llm.api_key if self.conversation else None,
self.conversation.state.agent.llm.api_key if self.conversation else None,
escapable=escapable
)
save_settings_confirmation()
@@ -163,14 +167,14 @@ class SettingsScreen:
model=model,
api_key=api_key,
base_url=base_url,
service_id="agent"
service_id="agent",
metadata=get_llm_metadata(model_name=model, llm_type="agent")
)
agent = self.agent_store.load()
if not agent:
agent = get_default_agent(
llm=llm,
working_dir=WORK_DIR,
cli_mode=True
)

View File

@@ -1,10 +1,14 @@
# openhands_cli/settings/store.py
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
from openhands.sdk import LocalFileStore, Agent, AgentContext
from openhands.sdk.preset.default import get_default_tools
from openhands_cli.locations import AGENT_SETTINGS_PATH, PERSISTENCE_DIR, WORK_DIR
from openhands.sdk.context.condenser import LLMSummarizingCondenser
from openhands.tools.preset.default import get_default_tools
from openhands_cli.llm_utils import get_llm_metadata
from openhands_cli.locations import AGENT_SETTINGS_PATH, MCP_CONFIG_FILE, PERSISTENCE_DIR, WORK_DIR
from prompt_toolkit import HTML, print_formatted_text
from fastmcp.mcp_config import MCPConfig
class AgentStore:
@@ -12,26 +16,57 @@ class AgentStore:
def __init__(self) -> None:
self.file_store = LocalFileStore(root=PERSISTENCE_DIR)
def load(self) -> Agent | None:
def load_mcp_configuration(self) -> dict[str, Any]:
try:
mcp_config_path = Path(self.file_store.root) / MCP_CONFIG_FILE
mcp_config = MCPConfig.from_file(mcp_config_path)
return mcp_config.to_dict()['mcpServers']
except Exception as e:
return {}
def load(self, session_id: str | None = None) -> Agent | None:
try:
str_spec = self.file_store.read(AGENT_SETTINGS_PATH)
agent = Agent.model_validate_json(str_spec)
# Update tools with most recent working directory
updated_tools = get_default_tools(
working_dir=WORK_DIR,
persistence_dir=PERSISTENCE_DIR,
enable_browser=False
)
# Create agent context with current working directory
agent_context = AgentContext(
system_message_suffix=f"You current working directory is: {WORK_DIR}",
)
additional_mcp_config = self.load_mcp_configuration()
mcp_config: dict = agent.mcp_config.copy().get('mcpServers', {})
mcp_config.update(additional_mcp_config)
# Update LLM metadata with current information
agent_llm_metadata = get_llm_metadata(
model_name=agent.llm.model,
llm_type="agent",
session_id=session_id
)
updated_llm = agent.llm.model_copy(update={"metadata": agent_llm_metadata})
condenser_updates = {}
if agent.condenser and isinstance(agent.condenser, LLMSummarizingCondenser):
condenser_updates["llm"] = agent.condenser.llm.model_copy(update={"metadata": get_llm_metadata(
model_name=agent.condenser.llm.model,
llm_type="condenser",
session_id=session_id
)})
agent = agent.model_copy(update={
"llm": updated_llm,
"tools": updated_tools,
"agent_context": agent_context
"mcp_config": {'mcpServers': mcp_config} if mcp_config else {},
"agent_context": agent_context,
"condenser": agent.condenser.model_copy(
update=condenser_updates
) if agent.condenser else None
})
return agent

View File

@@ -8,6 +8,7 @@ from prompt_toolkit.shortcuts import clear
from openhands_cli import __version__
from openhands_cli.pt_style import get_cli_style
from uuid import UUID
DEFAULT_STYLE = get_cli_style()
@@ -18,9 +19,9 @@ COMMANDS = {
"/clear": "Clear the screen",
"/status": "Display conversation details",
"/confirm": "Toggle confirmation mode on/off",
"/new": "Create a new conversation",
"/resume": "Resume a paused conversation",
"/settings": "Display and modify current settings",
"/mcp": "View MCP (Model Context Protocol) server configuration",
}
@@ -42,7 +43,7 @@ class CommandCompleter(Completer):
)
def display_banner(session_id: str) -> None:
def display_banner(conversation_id: str, resume: bool = False) -> None:
print_formatted_text(
HTML(r"""<gold>
___ _ _ _
@@ -58,7 +59,10 @@ def display_banner(session_id: str) -> None:
print_formatted_text(HTML(f"<grey>OpenHands CLI v{__version__}</grey>"))
print_formatted_text("")
print_formatted_text(HTML(f"<grey>Initialized conversation {session_id}</grey>"))
if not resume:
print_formatted_text(HTML(f"<grey>Initialized conversation {conversation_id}</grey>"))
else:
print_formatted_text(HTML(f"<grey>Resumed conversation {conversation_id}</grey>"))
print_formatted_text("")
@@ -80,10 +84,10 @@ def display_help() -> None:
print_formatted_text("")
def display_welcome(session_id: str = "chat") -> None:
def display_welcome(conversation_id: UUID, resume: bool = False) -> None:
"""Display welcome message."""
clear()
display_banner(session_id)
display_banner(str(conversation_id), resume)
print_formatted_text(HTML("<gold>Let's start building!</gold>"))
print_formatted_text(
HTML(

View File

@@ -1,6 +1,7 @@
from enum import Enum
from openhands_cli.tui.utils import StepCounter
from openhands_cli.user_actions.utils import NonEmptyValueValidator
from prompt_toolkit.completion import FuzzyWordCompleter
from pydantic import SecretStr
@@ -11,16 +12,6 @@ from openhands.sdk.llm import (
)
from openhands_cli.user_actions.utils import cli_confirm, cli_text_input
from prompt_toolkit.validation import Validator, ValidationError
class NonEmptyValueValidator(Validator):
def validate(self, document):
text = document.text
if not text:
raise ValidationError(
message="API key cannot be empty. Please enter a valid API key."
)
class SettingsType(Enum):
@@ -36,7 +27,7 @@ def settings_type_confirmation() -> SettingsType:
'Go back',
]
index = cli_confirm(question, choices)
index = cli_confirm(question, choices, escapable=True)
if choices[index] == 'Go back':
raise KeyboardInterrupt
@@ -150,7 +141,7 @@ def save_settings_confirmation() -> bool:
discard = 'No, discard'
options = ['Yes, save', discard]
index = cli_confirm(question, options)
index = cli_confirm(question, options, escapable=True)
if options[index] == discard:
raise KeyboardInterrupt

View File

@@ -4,7 +4,6 @@ from typing import Optional
from pydantic import BaseModel
from openhands.sdk.security.confirmation_policy import ConfirmationPolicyBase
class UserConfirmation(Enum):
ACCEPT = "accept"
REJECT = "reject"

View File

@@ -1,3 +1,5 @@
from openhands_cli.tui.tui import CommandCompleter
from prompt_toolkit import HTML, PromptSession
from prompt_toolkit.application import Application
from prompt_toolkit.completion import Completer
from prompt_toolkit.input.base import Input
@@ -9,7 +11,10 @@ from prompt_toolkit.layout.dimension import Dimension
from prompt_toolkit.layout.layout import Layout
from prompt_toolkit.output.base import Output
from prompt_toolkit.shortcuts import prompt
from prompt_toolkit.validation import Validator
from prompt_toolkit.validation import Validator, ValidationError
from prompt_toolkit.styles import Style
from prompt_toolkit.styles import merge_styles
from openhands_cli.tui import DEFAULT_STYLE
@@ -128,13 +133,17 @@ def cli_text_input(
@kb.add('c-c')
def _(event: KeyPressEvent) -> None:
raise KeyboardInterrupt()
event.app.exit(exception=KeyboardInterrupt())
@kb.add('c-p')
def _(event: KeyPressEvent) -> None:
raise KeyboardInterrupt()
event.app.exit(exception=KeyboardInterrupt())
@kb.add("enter")
def _handle_enter(event: KeyPressEvent):
event.app.exit(result=event.current_buffer.text)
reason = str(
prompt(
question,
@@ -146,3 +155,51 @@ def cli_text_input(
)
)
return reason.strip()
def get_session_prompter(
input: Input | None = None, # strictly for unit testing
output: Output | None = None, # strictly for unit testing
) -> PromptSession:
bindings = KeyBindings()
@bindings.add("\\", "enter")
def _(event: KeyPressEvent) -> None:
# Typing '\' + Enter forces a newline regardless
event.current_buffer.insert_text("\n")
@bindings.add("enter")
def _handle_enter(event: KeyPressEvent):
event.app.exit(result=event.current_buffer.text)
@bindings.add("c-c")
def _keyboard_interrupt(event: KeyPressEvent):
event.app.exit(exception=KeyboardInterrupt())
session = PromptSession(
completer=CommandCompleter(),
key_bindings=bindings,
prompt_continuation=lambda width, line_number, is_soft_wrap: "...",
multiline=True,
input=input,
output=output,
style=DEFAULT_STYLE,
placeholder=HTML(
"<placeholder>"
"Type your message… (tip: press <b>\\</b> + <b>Enter</b> to insert a newline)"
"</placeholder>"
),
)
return session
class NonEmptyValueValidator(Validator):
def validate(self, document):
text = document.text
if not text:
raise ValidationError(
message="API key cannot be empty. Please enter a valid API key."
)

View File

@@ -82,5 +82,5 @@ disallow_untyped_defs = true
ignore_missing_imports = true
[tool.uv.sources]
openhands-sdk = { git = "https://github.com/All-Hands-AI/agent-sdk.git", subdirectory = "openhands/sdk", rev = "f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab" }
openhands-tools = { git = "https://github.com/All-Hands-AI/agent-sdk.git", subdirectory = "openhands/tools", rev = "f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab" }
openhands-sdk = { git = "https://github.com/All-Hands-AI/agent-sdk.git", subdirectory = "openhands/sdk", rev = "711efcbadaa78a0b6b20699976e495ddf995767f" }
openhands-tools = { git = "https://github.com/All-Hands-AI/agent-sdk.git", subdirectory = "openhands/tools", rev = "711efcbadaa78a0b6b20699976e495ddf995767f" }

View File

@@ -6,10 +6,10 @@ Tests for confirmation mode functionality in OpenHands CLI.
import os
from concurrent.futures import ThreadPoolExecutor
from typing import Any
from unittest.mock import MagicMock, patch
from unittest.mock import ANY, MagicMock, patch
import pytest
from openhands.sdk import ActionBase
from openhands.sdk import ActionBase, Workspace
from openhands.sdk.security.confirmation_policy import AlwaysConfirm, NeverConfirm, ConfirmRisky, SecurityRisk
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.output.defaults import DummyOutput
@@ -38,7 +38,12 @@ class TestConfirmationMode:
patch('openhands_cli.setup.AgentStore') as mock_agent_store_class,
patch('openhands_cli.setup.print_formatted_text') as mock_print,
patch('openhands_cli.setup.HTML') as mock_html,
patch('openhands_cli.setup.uuid') as mock_uuid,
):
# Mock dependencies
mock_conversation_id = MagicMock()
mock_uuid.uuid4.return_value = mock_conversation_id
# Mock AgentStore
mock_agent_store_instance = MagicMock()
mock_agent_instance = MagicMock()
@@ -56,7 +61,12 @@ class TestConfirmationMode:
assert result == mock_conversation_instance
mock_agent_store_class.assert_called_once()
mock_agent_store_instance.load.assert_called_once()
mock_conversation_class.assert_called_once_with(agent=mock_agent_instance)
mock_conversation_class.assert_called_once_with(
agent=mock_agent_instance,
workspace=ANY,
persistence_dir=ANY,
conversation_id=mock_conversation_id
)
# Verify print_formatted_text was called
mock_print.assert_called_once()
@@ -73,7 +83,7 @@ class TestConfirmationMode:
# Should raise MissingAgentSpec
with pytest.raises(MissingAgentSpec) as exc_info:
setup_conversation()
assert "Agent specification not found" in str(exc_info.value)
mock_agent_store_class.assert_called_once()
mock_agent_store_instance.load.assert_called_once()
@@ -363,7 +373,7 @@ class TestConfirmationMode:
assert runner.is_confirmation_mode_enabled is True
# Mock get_unmatched_actions to return some actions
with patch('openhands_cli.runner.get_unmatched_actions') as mock_get_actions:
with patch('openhands_cli.runner.ConversationState.get_unmatched_actions') as mock_get_actions:
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo test'
@@ -414,7 +424,7 @@ class TestConfirmationMode:
assert runner.is_confirmation_mode_enabled is True
# Mock get_unmatched_actions to return some actions
with patch('openhands_cli.runner.get_unmatched_actions') as mock_get_actions:
with patch('openhands_cli.runner.ConversationState.get_unmatched_actions') as mock_get_actions:
mock_action = MagicMock()
mock_action.tool_name = 'bash'
mock_action.action = 'echo test'

View File

@@ -5,7 +5,7 @@ from unittest.mock import patch, MagicMock
from openhands.sdk import Agent, LLM, ToolSpec
from openhands_cli.locations import WORK_DIR, PERSISTENCE_DIR
from openhands_cli.tui.settings.store import AgentStore
from openhands.sdk.preset.default import get_default_tools
from openhands.tools.preset.default import get_default_tools
class TestDirectorySeparation:
@@ -39,14 +39,14 @@ class TestToolSpecFix:
mock_agent = Agent(
llm=LLM(model="test/model", api_key="test-key", service_id="test-service"),
tools=[
ToolSpec(name="BashTool", params={"working_dir": original_working_dir}),
ToolSpec(name="FileEditorTool", params={"workspace_root": original_working_dir}),
ToolSpec(name="TaskTrackerTool", params={"save_dir": "value"}),
ToolSpec(name="BashTool"),
ToolSpec(name="FileEditorTool"),
ToolSpec(name="TaskTrackerTool"),
]
)
# Mock the file store to return our test agent
with patch('openhands_cli.tui.settings.store.LocalFileStore') as mock_file_store:
with patch("openhands_cli.tui.settings.store.LocalFileStore") as mock_file_store:
mock_store_instance = MagicMock()
mock_file_store.return_value = mock_store_instance
mock_store_instance.read.return_value = mock_agent.model_dump_json()
@@ -64,14 +64,3 @@ class TestToolSpecFix:
assert "BashTool" in tool_names
assert "FileEditorTool" in tool_names
assert "TaskTrackerTool" in tool_names
for tool_spec in loaded_agent.tools:
if tool_spec.name == "BashTool":
assert tool_spec.params["working_dir"] == WORK_DIR
assert tool_spec.params["working_dir"] != original_working_dir
elif tool_spec.name == "FileEditorTool":
assert tool_spec.params["workspace_root"] == WORK_DIR
assert tool_spec.params["workspace_root"] != original_working_dir
elif tool_spec.name == "TaskTrackerTool":
# TaskTrackerTool should use WORK_DIR/.openhands_tasks
assert tool_spec.params["save_dir"] == PERSISTENCE_DIR

View File

@@ -0,0 +1,69 @@
#!/usr/bin/env python3
"""
Unit tests for the loading animation functionality.
"""
import threading
import time
import unittest
from unittest.mock import patch
from openhands_cli.listeners.loading_listener import (
LoadingContext,
display_initialization_animation
)
class TestLoadingAnimation(unittest.TestCase):
"""Test cases for loading animation functionality."""
def test_loading_context_manager(self):
"""Test that LoadingContext works as a context manager."""
with LoadingContext("Test loading...") as ctx:
self.assertIsInstance(ctx, LoadingContext)
self.assertEqual(ctx.text, "Test loading...")
self.assertIsInstance(ctx.is_loaded, threading.Event)
self.assertIsNotNone(ctx.loading_thread)
# Give the thread a moment to start
time.sleep(0.1)
self.assertTrue(ctx.loading_thread.is_alive())
# After exiting context, thread should be stopped
time.sleep(0.1)
self.assertFalse(ctx.loading_thread.is_alive())
@patch('sys.stdout')
def test_animation_writes_while_running_and_stops_after(self, mock_stdout):
"""Ensure stdout is written while animation runs and stops after it ends."""
is_loaded = threading.Event()
animation_thread = threading.Thread(
target=display_initialization_animation,
args=("Test output", is_loaded),
daemon=True
)
animation_thread.start()
# Let it run a bit and check calls
time.sleep(0.2)
calls_while_running = mock_stdout.write.call_count
self.assertGreater(calls_while_running, 0, "Expected writes while spinner runs")
# Stop animation
is_loaded.set()
time.sleep(0.2)
animation_thread.join(timeout=1.0)
calls_after_stop = mock_stdout.write.call_count
# Wait a moment to detect any stray writes after thread finished
time.sleep(0.2)
self.assertEqual(
calls_after_stop,
mock_stdout.write.call_count,
"No extra writes should occur after animation stops"
)
if __name__ == "__main__":
unittest.main()

View File

@@ -1,5 +1,6 @@
"""Tests for main entry point functionality."""
import uuid
from unittest.mock import MagicMock, patch
import pytest
@@ -10,27 +11,23 @@ from openhands_cli import simple_main
class TestMainEntryPoint:
"""Test the main entry point behavior."""
@patch('openhands_cli.agent_chat.setup_conversation')
@patch('openhands_cli.agent_chat.ConversationRunner')
@patch('openhands_cli.agent_chat.PromptSession')
@patch('openhands_cli.simple_main.run_cli_entry')
@patch('sys.argv', ['openhands-cli'])
def test_main_starts_agent_chat_directly(
self, mock_prompt_session: MagicMock, mock_runner: MagicMock, mock_setup_conversation: MagicMock
self, mock_run_agent_chat: MagicMock
) -> None:
"""Test that main() starts agent chat directly when setup succeeds."""
# Mock setup_conversation to return a valid conversation
mock_conversation = MagicMock()
mock_setup_conversation.return_value = mock_conversation
# Mock prompt session to raise KeyboardInterrupt to exit the loop
mock_prompt_session.return_value.prompt.side_effect = KeyboardInterrupt()
# Mock run_cli_entry to raise KeyboardInterrupt to exit gracefully
mock_run_agent_chat.side_effect = KeyboardInterrupt()
# Should complete without raising an exception (graceful exit)
simple_main.main()
# Should call setup_conversation
mock_setup_conversation.assert_called_once()
# Should call run_cli_entry with no resume conversation ID
mock_run_agent_chat.assert_called_once_with(resume_conversation_id=None)
@patch('openhands_cli.simple_main.run_cli_entry')
@patch('sys.argv', ['openhands-cli'])
def test_main_handles_import_error(self, mock_run_agent_chat: MagicMock) -> None:
"""Test that main() handles ImportError gracefully."""
mock_run_agent_chat.side_effect = ImportError('Missing dependency')
@@ -41,41 +38,32 @@ class TestMainEntryPoint:
assert str(exc_info.value) == 'Missing dependency'
@patch('openhands_cli.agent_chat.setup_conversation')
@patch('openhands_cli.agent_chat.ConversationRunner')
@patch('openhands_cli.agent_chat.PromptSession')
@patch('openhands_cli.simple_main.run_cli_entry')
@patch('sys.argv', ['openhands-cli'])
def test_main_handles_keyboard_interrupt(
self, mock_prompt_session: MagicMock, mock_runner: MagicMock, mock_setup_conversation: MagicMock
self, mock_run_agent_chat: MagicMock
) -> None:
"""Test that main() handles KeyboardInterrupt gracefully."""
# Mock setup_conversation to return a valid conversation
mock_conversation = MagicMock()
mock_setup_conversation.return_value = mock_conversation
# Mock prompt session to raise KeyboardInterrupt
mock_prompt_session.return_value.prompt.side_effect = KeyboardInterrupt()
# Should complete without raising an exception (graceful exit)
simple_main.main()
@patch('openhands_cli.agent_chat.setup_conversation')
@patch('openhands_cli.agent_chat.ConversationRunner')
@patch('openhands_cli.agent_chat.PromptSession')
def test_main_handles_eof_error(
self, mock_prompt_session: MagicMock, mock_runner: MagicMock, mock_setup_conversation: MagicMock
) -> None:
"""Test that main() handles EOFError gracefully."""
# Mock setup_conversation to return a valid conversation
mock_conversation = MagicMock()
mock_setup_conversation.return_value = mock_conversation
# Mock prompt session to raise EOFError
mock_prompt_session.return_value.prompt.side_effect = EOFError()
# Mock run_cli_entry to raise KeyboardInterrupt
mock_run_agent_chat.side_effect = KeyboardInterrupt()
# Should complete without raising an exception (graceful exit)
simple_main.main()
@patch('openhands_cli.simple_main.run_cli_entry')
@patch('sys.argv', ['openhands-cli'])
def test_main_handles_eof_error(
self, mock_run_agent_chat: MagicMock
) -> None:
"""Test that main() handles EOFError gracefully."""
# Mock run_cli_entry to raise EOFError
mock_run_agent_chat.side_effect = EOFError()
# Should complete without raising an exception (graceful exit)
simple_main.main()
@patch('openhands_cli.simple_main.run_cli_entry')
@patch('sys.argv', ['openhands-cli'])
def test_main_handles_general_exception(
self, mock_run_agent_chat: MagicMock
) -> None:
@@ -87,3 +75,18 @@ class TestMainEntryPoint:
simple_main.main()
assert str(exc_info.value) == 'Unexpected error'
@patch('openhands_cli.simple_main.run_cli_entry')
@patch('sys.argv', ['openhands-cli', '--resume', 'test-conversation-id'])
def test_main_with_resume_argument(
self, mock_run_agent_chat: MagicMock
) -> None:
"""Test that main() passes resume conversation ID when provided."""
# Mock run_cli_entry to raise KeyboardInterrupt to exit gracefully
mock_run_agent_chat.side_effect = KeyboardInterrupt()
# Should complete without raising an exception (graceful exit)
simple_main.main()
# Should call run_cli_entry with the provided resume conversation ID
mock_run_agent_chat.assert_called_once_with(resume_conversation_id='test-conversation-id')

View File

@@ -0,0 +1,199 @@
"""Parametrized tests for MCP configuration screen functionality."""
import json
from pathlib import Path
from unittest.mock import patch
import pytest
from openhands_cli.locations import MCP_CONFIG_FILE
from openhands_cli.tui.settings.mcp_screen import MCPScreen
from openhands.sdk import Agent, LLM
@pytest.fixture
def persistence_dir(tmp_path, monkeypatch):
"""Patch PERSISTENCE_DIR to tmp and return the directory Path."""
monkeypatch.setattr(
"openhands_cli.tui.settings.mcp_screen.PERSISTENCE_DIR",
str(tmp_path),
raising=True,
)
return tmp_path
def _create_agent(mcp_config=None) -> Agent:
if mcp_config is None:
mcp_config = {}
return Agent(
llm=LLM(model="test-model", api_key="test-key", service_id="test-service"),
tools=[],
mcp_config=mcp_config,
)
def _maybe_write_mcp_file(dirpath: Path, file_content):
"""Write mcp.json if file_content is provided.
file_content:
- None -> do not create file (missing)
- "INVALID"-> write invalid JSON
- dict -> dump as JSON
"""
if file_content is None:
return
cfg_path = dirpath / MCP_CONFIG_FILE
if file_content == "INVALID":
cfg_path.write_text('{"invalid": json content}')
else:
cfg_path.write_text(json.dumps(file_content))
# Shared "always expected" help text snippets
ALWAYS_EXPECTED = [
"MCP (Model Context Protocol) Configuration",
"To get started:",
"~/.openhands/mcp.json",
"https://gofastmcp.com/clients/client#configuration-format",
"Restart your OpenHands session",
]
CASES = [
# Agent has an existing server; should list "Current Agent MCP Servers"
dict(
id="agent_has_existing",
agent_mcp={
"mcpServers": {
"existing_server": {"command": "python", "args": ["-m", "existing_server"]}
}
},
file_content=None, # no incoming file
expected=[
"Current Agent MCP Servers:",
"existing_server",
],
unexpected=[],
),
# Agent has none; should show "None configured on the current agent"
dict(
id="agent_has_none",
agent_mcp={},
file_content=None,
expected=[
"Current Agent MCP Servers:",
"None configured on the current agent",
],
unexpected=[],
),
# New servers present only in mcp.json
dict(
id="new_servers_on_restart",
agent_mcp={},
file_content={
"mcpServers": {
"fetch": {"command": "uvx", "args": ["mcp-server-fetch"]},
"notion": {"url": "https://mcp.notion.com/mcp", "auth": "oauth"},
}
},
expected=[
"Incoming Servers on Restart",
"New servers (will be added):",
"fetch",
"notion",
],
unexpected=[],
),
# Overriding/updating servers present in both agent and mcp.json (but different config)
dict(
id="overriding_servers_on_restart",
agent_mcp={
"mcpServers": {
"fetch": {"command": "python", "args": ["-m", "old_fetch_server"]}
}
},
file_content={
"mcpServers": {"fetch": {"command": "uvx", "args": ["mcp-server-fetch"]}}
},
expected=[
"Incoming Servers on Restart",
"Updated servers (configuration will change):",
"fetch",
"Current:",
"Incoming:",
],
unexpected=[],
),
# All servers already synced (matching config)
dict(
id="already_synced",
agent_mcp={
"mcpServers": {
"fetch": {
"command": "uvx",
"args": ["mcp-server-fetch"],
"env": {},
"transport": "stdio",
}
}
},
file_content={
"mcpServers": {"fetch": {"command": "uvx", "args": ["mcp-server-fetch"]}}
},
expected=[
"Incoming Servers on Restart",
"All configured servers match the current agent configuration",
],
unexpected=[],
),
# Invalid JSON file handling
dict(
id="invalid_json_file",
agent_mcp={},
file_content="INVALID",
expected=[
"Invalid MCP configuration file",
"Please check your configuration file format",
],
unexpected=[],
),
# Missing JSON file handling
dict(
id="missing_json_file",
agent_mcp={},
file_content=None, # explicitly missing
expected=[
"Configuration file not found",
"No incoming servers detected for next restart",
],
unexpected=[],
),
]
@pytest.mark.parametrize("case", CASES, ids=[c["id"] for c in CASES])
@patch("openhands_cli.tui.settings.mcp_screen.print_formatted_text")
def test_display_mcp_info_parametrized(mock_print, case, persistence_dir):
"""Table-driven test for MCPScreen.display_mcp_info covering all scenarios."""
# Arrange
agent = _create_agent(case["agent_mcp"])
_maybe_write_mcp_file(persistence_dir, case["file_content"])
screen = MCPScreen()
# Act
screen.display_mcp_info(agent)
# Gather output
all_calls = [str(call_args) for call_args in mock_print.call_args_list]
content = " ".join(all_calls)
# Invariants: help instructions should always be present
for snippet in ALWAYS_EXPECTED:
assert snippet in content, f"Missing help snippet: {snippet}"
# Scenario-specific expectations
for snippet in case["expected"]:
assert snippet in content, f"Expected snippet not found for case {case['id']}: {snippet}"
for snippet in case.get("unexpected", []):
assert snippet not in content, f"Unexpected snippet found for case {case['id']}: {snippet}"

View File

@@ -0,0 +1,102 @@
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Optional, Type
import pytest
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.input.defaults import create_pipe_input
from prompt_toolkit.output.defaults import DummyOutput
from openhands_cli.user_actions.utils import get_session_prompter
from tests.utils import _send_keys
def _run_prompt_and_type(
prompt_text: str,
keys: str,
*,
expect_exception: Optional[Type[BaseException]] = None,
timeout: float = 2.0,
settle: float = 0.05,
) -> str | None:
"""
Helper to:
1) create a pipe + session,
2) start session.prompt in a background thread,
3) send keys, and
4) return the result or raise the expected exception.
Returns:
- The prompt result (str) if no exception expected.
- None if an exception is expected and raised.
"""
with create_pipe_input() as pipe:
session = get_session_prompter(input=pipe, output=DummyOutput())
with ThreadPoolExecutor(max_workers=1) as ex:
fut = ex.submit(session.prompt, HTML(prompt_text))
# Allow the prompt loop to start consuming input
time.sleep(settle)
_send_keys(pipe, keys)
if expect_exception:
with pytest.raises(expect_exception):
fut.result(timeout=timeout)
return None
return fut.result(timeout=timeout)
@pytest.mark.parametrize(
"desc,keys,expected",
[
("basic single line", "hello world\r", "hello world"),
("empty input", "\r", ""),
("single multiline via backslash-enter", "line 1\\\rline 2\r", "line 1\nline 2"),
(
"multiple multiline segments",
"first line\\\rsecond line\\\rthird line\r",
"first line\nsecond line\nthird line",
),
(
"backslash-only newline then text",
"\\\rafter newline\r",
"\nafter newline",
),
(
"mixed content (code-like)",
"def function():\\\r return 'hello'\\\r # end of function\r",
"def function():\n return 'hello'\n # end of function",
),
(
"whitespace preservation (including blank line)",
" indented line\\\r\\\r more indented\r",
" indented line\n\n more indented",
),
(
"special characters",
"echo 'hello world'\\\rgrep -n \"pattern\" file.txt\r",
"echo 'hello world'\ngrep -n \"pattern\" file.txt",
),
],
)
def test_get_session_prompter_scenarios(desc, keys, expected):
"""Covers most behaviors via parametrization to reduce duplication."""
result = _run_prompt_and_type("<gold>> </gold>", keys)
assert result == expected
def test_get_session_prompter_keyboard_interrupt():
"""Focused test for Ctrl+C behavior."""
_run_prompt_and_type("<gold>> </gold>", "\x03", expect_exception=KeyboardInterrupt)
def test_get_session_prompter_default_parameters():
"""Lightweight sanity check for default construction."""
session = get_session_prompter()
assert session is not None
assert session.multiline is True
assert session.key_bindings is not None
assert session.completer is not None
# Prompt continuation should be callable and return the expected string
cont = session.prompt_continuation
assert callable(cont)
assert cont(80, 1, False) == "..."

View File

@@ -4,7 +4,7 @@ from openhands_cli.tui.settings.settings_screen import SettingsScreen
from pathlib import Path
from openhands.sdk import LLM, Conversation, LocalFileStore
from openhands.sdk.preset.default import get_default_agent
from openhands.tools.preset.default import get_default_agent
from openhands_cli.tui.settings.store import AgentStore
from openhands_cli.user_actions.settings_action import SettingsType
from pydantic import SecretStr
@@ -26,8 +26,7 @@ def seed_file(path: Path, model: str = "openai/gpt-4o-mini", api_key: str = "sk-
store = AgentStore()
store.file_store = LocalFileStore(root=str(path))
agent = get_default_agent(
llm=LLM(model=model, api_key=SecretStr(api_key), service_id="test-service"),
working_dir=str(path)
llm=LLM(model=model, api_key=SecretStr(api_key), service_id="test-service")
)
store.save(agent)

View File

@@ -80,9 +80,9 @@ def test_commands_dict() -> None:
'/clear',
'/status',
'/confirm',
'/new',
'/resume',
'/settings',
'/mcp',
}
assert set(COMMANDS.keys()) == expected_commands

8
openhands-cli/uv.lock generated
View File

@@ -1484,8 +1484,8 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "openhands-sdk", git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Fsdk&rev=f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab" },
{ name = "openhands-tools", git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Ftools&rev=f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab" },
{ name = "openhands-sdk", git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Fsdk&rev=711efcbadaa78a0b6b20699976e495ddf995767f" },
{ name = "openhands-tools", git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Ftools&rev=711efcbadaa78a0b6b20699976e495ddf995767f" },
{ name = "prompt-toolkit", specifier = ">=3" },
{ name = "typer", specifier = ">=0.17.4" },
]
@@ -1505,7 +1505,7 @@ dev = [
[[package]]
name = "openhands-sdk"
version = "1.0.0"
source = { git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Fsdk&rev=f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab#f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab" }
source = { git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Fsdk&rev=711efcbadaa78a0b6b20699976e495ddf995767f#711efcbadaa78a0b6b20699976e495ddf995767f" }
dependencies = [
{ name = "fastmcp" },
{ name = "litellm" },
@@ -1519,7 +1519,7 @@ dependencies = [
[[package]]
name = "openhands-tools"
version = "1.0.0"
source = { git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Ftools&rev=f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab#f0b9bcb5de574f5c4fdc8e1c153bbdd0bf1216ab" }
source = { git = "https://github.com/All-Hands-AI/agent-sdk.git?subdirectory=openhands%2Ftools&rev=711efcbadaa78a0b6b20699976e495ddf995767f#711efcbadaa78a0b6b20699976e495ddf995767f" }
dependencies = [
{ name = "bashlex" },
{ name = "binaryornot" },