mirror of
https://github.com/All-Hands-AI/OpenHands.git
synced 2026-04-29 03:00:45 -04:00
Compare commits
11 Commits
openhands/
...
v1
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
953f99a147 | ||
|
|
1d78513407 | ||
|
|
d51c6bb992 | ||
|
|
1cd8eada2b | ||
|
|
44c4e0e5fd | ||
|
|
a9982f96c6 | ||
|
|
7112b4e329 | ||
|
|
c2d1d15a8f | ||
|
|
d2bb882c96 | ||
|
|
e995882194 | ||
|
|
ef1441bbe5 |
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
from openhands_cli.listeners.pause_listener import PauseListener
|
||||
from openhands_cli.listeners.loading_listener import LoadingContext
|
||||
|
||||
__all__ = [
|
||||
"PauseListener",
|
||||
"LoadingContext"
|
||||
]
|
||||
|
||||
61
openhands-cli/openhands_cli/listeners/loading_listener.py
Normal file
61
openhands-cli/openhands_cli/listeners/loading_listener.py
Normal 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
|
||||
@@ -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)
|
||||
|
||||
55
openhands-cli/openhands_cli/llm_utils.py
Normal file
55
openhands-cli/openhands_cli/llm_utils.py
Normal 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
|
||||
@@ -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"
|
||||
|
||||
@@ -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])
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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>")
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
202
openhands-cli/openhands_cli/tui/settings/mcp_screen.py
Normal file
202
openhands-cli/openhands_cli/tui/settings/mcp_screen.py
Normal 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)
|
||||
@@ -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
|
||||
)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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."
|
||||
)
|
||||
|
||||
@@ -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" }
|
||||
|
||||
@@ -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'
|
||||
|
||||
@@ -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
|
||||
|
||||
69
openhands-cli/tests/test_loading.py
Normal file
69
openhands-cli/tests/test_loading.py
Normal 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()
|
||||
@@ -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')
|
||||
|
||||
199
openhands-cli/tests/test_mcp_config_validation.py
Normal file
199
openhands-cli/tests/test_mcp_config_validation.py
Normal 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}"
|
||||
102
openhands-cli/tests/test_session_prompter.py
Normal file
102
openhands-cli/tests/test_session_prompter.py
Normal 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) == "..."
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
8
openhands-cli/uv.lock
generated
@@ -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" },
|
||||
|
||||
Reference in New Issue
Block a user