4
4
including command execution, script running, and process management.
5
5
"""
6
6
7
- import json
8
- import re
9
7
from abc import ABC , abstractmethod
10
8
from enum import Enum
11
9
from typing import Any , Self , final
12
10
13
11
from fastmcp import Context as MCPContext
14
- from pydantic import BaseModel
15
12
16
13
from mcp_claude_code .tools .common .base import BaseTool
17
14
from mcp_claude_code .tools .common .permissions import PermissionManager
@@ -26,80 +23,6 @@ class BashCommandStatus(Enum):
26
23
HARD_TIMEOUT = "hard_timeout"
27
24
28
25
29
- # PS1 metadata constants
30
- CMD_OUTPUT_PS1_BEGIN = "\n ###PS1JSON###\n "
31
- CMD_OUTPUT_PS1_END = "\n ###PS1END###"
32
- CMD_OUTPUT_METADATA_PS1_REGEX = re .compile (
33
- f"^{ CMD_OUTPUT_PS1_BEGIN .strip ()} (.*?){ CMD_OUTPUT_PS1_END .strip ()} " ,
34
- re .DOTALL | re .MULTILINE ,
35
- )
36
-
37
-
38
- class CmdOutputMetadata (BaseModel ):
39
- """Rich metadata captured from PS1 prompts and command execution."""
40
-
41
- exit_code : int = - 1
42
- pid : int = - 1
43
- username : str | None = None
44
- hostname : str | None = None
45
- working_dir : str | None = None
46
- py_interpreter_path : str | None = None
47
- prefix : str = "" # Prefix to add to command output
48
- suffix : str = "" # Suffix to add to command output
49
-
50
- @classmethod
51
- def to_ps1_prompt (cls ) -> str :
52
- """Convert the required metadata into a PS1 prompt."""
53
- prompt = CMD_OUTPUT_PS1_BEGIN
54
- json_str = json .dumps (
55
- {
56
- "pid" : "$!" ,
57
- "exit_code" : "$?" ,
58
- "username" : r"\u" ,
59
- "hostname" : r"\h" ,
60
- "working_dir" : r"$(pwd)" ,
61
- "py_interpreter_path" : r'$(which python 2>/dev/null || echo "")' ,
62
- },
63
- indent = 2 ,
64
- )
65
- # Make sure we escape double quotes in the JSON string
66
- # So that PS1 will keep them as part of the output
67
- prompt += json_str .replace ('"' , r"\"" )
68
- prompt += CMD_OUTPUT_PS1_END + "\n " # Ensure there's a newline at the end
69
- return prompt
70
-
71
- @classmethod
72
- def matches_ps1_metadata (cls , string : str ) -> list [re .Match [str ]]:
73
- """Find all PS1 metadata matches in a string."""
74
- matches = []
75
- for match in CMD_OUTPUT_METADATA_PS1_REGEX .finditer (string ):
76
- try :
77
- json .loads (match .group (1 ).strip ()) # Try to parse as JSON
78
- matches .append (match )
79
- except json .JSONDecodeError :
80
- continue # Skip if not valid JSON
81
- return matches
82
-
83
- @classmethod
84
- def from_ps1_match (cls , match : re .Match [str ]) -> Self :
85
- """Extract the required metadata from a PS1 prompt."""
86
- metadata = json .loads (match .group (1 ))
87
- # Create a copy of metadata to avoid modifying the original
88
- processed = metadata .copy ()
89
- # Convert numeric fields
90
- if "pid" in metadata :
91
- try :
92
- processed ["pid" ] = int (float (str (metadata ["pid" ])))
93
- except (ValueError , TypeError ):
94
- processed ["pid" ] = - 1
95
- if "exit_code" in metadata :
96
- try :
97
- processed ["exit_code" ] = int (float (str (metadata ["exit_code" ])))
98
- except (ValueError , TypeError ):
99
- processed ["exit_code" ] = - 1
100
- return cls (** processed )
101
-
102
-
103
26
@final
104
27
class CommandResult :
105
28
"""Represents the result of a command execution with rich metadata."""
@@ -111,7 +34,6 @@ def __init__(
111
34
stderr : str = "" ,
112
35
error_message : str | None = None ,
113
36
session_id : str | None = None ,
114
- metadata : CmdOutputMetadata | None = None ,
115
37
status : BashCommandStatus = BashCommandStatus .COMPLETED ,
116
38
command : str = "" ,
117
39
):
@@ -123,7 +45,6 @@ def __init__(
123
45
stderr: Standard error from the command
124
46
error_message: Optional error message for failure cases
125
47
session_id: Optional session ID used for the command execution
126
- metadata: Rich metadata from command execution
127
48
status: Command execution status
128
49
command: The original command that was executed
129
50
"""
@@ -132,7 +53,6 @@ def __init__(
132
53
self .stderr : str = stderr
133
54
self .error_message : str | None = error_message
134
55
self .session_id : str | None = session_id
135
- self .metadata : CmdOutputMetadata = metadata or CmdOutputMetadata ()
136
56
self .status : BashCommandStatus = status
137
57
self .command : str = command
138
58
@@ -180,7 +100,7 @@ def message(self) -> str:
180
100
return f"Command `{ self .command } ` executed with exit code { self .return_code } ."
181
101
182
102
def format_output (self , include_exit_code : bool = True ) -> str :
183
- """Format the command output as a string with rich metadata .
103
+ """Format the command output as a string.
184
104
185
105
Args:
186
106
include_exit_code: Whether to include the exit code in the output
@@ -206,26 +126,9 @@ def format_output(self, include_exit_code: bool = True) -> str:
206
126
if include_exit_code and (self .return_code != 0 or not self .error_message ):
207
127
result_parts .append (f"Exit code: { self .return_code } " )
208
128
209
- # Add working directory if available
210
- if self .metadata .working_dir :
211
- result_parts .append (f"Working directory: { self .metadata .working_dir } " )
212
-
213
- # Add Python interpreter if available
214
- if self .metadata .py_interpreter_path :
215
- result_parts .append (
216
- f"Python interpreter: { self .metadata .py_interpreter_path } "
217
- )
218
-
219
- # Format the main output with prefix and suffix
220
- output_content = self .stdout
221
- if self .metadata .prefix :
222
- output_content = self .metadata .prefix + output_content
223
- if self .metadata .suffix :
224
- output_content = output_content + self .metadata .suffix
225
-
226
129
# Add stdout if present
227
- if output_content :
228
- result_parts .append (f"STDOUT:\n { output_content } " )
130
+ if self . stdout :
131
+ result_parts .append (f"STDOUT:\n { self . stdout } " )
229
132
230
133
# Add stderr if present
231
134
if self .stderr :
@@ -235,26 +138,10 @@ def format_output(self, include_exit_code: bool = True) -> str:
235
138
return "\n \n " .join (result_parts )
236
139
237
140
def to_agent_observation (self ) -> str :
238
- """Format the result for agent consumption (similar to OpenHands) ."""
141
+ """Format the result for agent consumption."""
239
142
content = self .stdout
240
- if self .metadata .prefix :
241
- content = self .metadata .prefix + content
242
- if self .metadata .suffix :
243
- content = content + self .metadata .suffix
244
143
245
144
additional_info : list [str ] = []
246
- if self .metadata .working_dir :
247
- additional_info .append (
248
- f"[Current working directory: { self .metadata .working_dir } ]"
249
- )
250
- if self .metadata .py_interpreter_path :
251
- additional_info .append (
252
- f"[Python interpreter: { self .metadata .py_interpreter_path } ]"
253
- )
254
- if self .metadata .exit_code != - 1 :
255
- additional_info .append (
256
- f"[Command finished with exit code { self .metadata .exit_code } ]"
257
- )
258
145
if self .session_id :
259
146
additional_info .append (f"[Session ID: { self .session_id } ]" )
260
147
0 commit comments