MCPcopy Create free account
hub / github.com/IBM/mcp-cli / DummyToolManager

Class DummyToolManager

tests/chat/test_tool_processor.py:86–170  ·  view source on GitHub ↗

Mock tool manager with execute_tool and stream_execute_tools methods.

Source from the content-addressed store, hash-verified

84
85
86class DummyToolManager:
87 """Mock tool manager with execute_tool and stream_execute_tools methods."""
88
89 def __init__(self, return_result=None, raise_exception=False):
90 self.return_result = return_result or {
91 "isError": False,
92 "content": "Tool executed successfully",
93 }
94 self.raise_exception = raise_exception
95 self.executed_tool = None
96 self.executed_args = None
97
98 async def execute_tool(self, tool_name, arguments, namespace=None, timeout=None):
99 self.executed_tool = tool_name
100 self.executed_args = arguments
101 if self.raise_exception:
102 raise Exception("Simulated execute_tool exception")
103
104 # Return a ToolCallResult object, not a dict
105 if self.return_result.get("isError"):
106 return ToolCallResult(
107 tool_name=tool_name,
108 success=False,
109 result=None,
110 error=self.return_result.get("error", "Simulated error"),
111 )
112 else:
113 return ToolCallResult(
114 tool_name=tool_name,
115 success=True,
116 result=self.return_result.get("content"),
117 error=None,
118 )
119
120 async def stream_execute_tools(
121 self, calls, timeout=None, on_tool_start=None, max_concurrency=4
122 ):
123 """Yield CTPToolResult for each call."""
124 import platform
125 import os
126
127 for call in calls:
128 self.executed_tool = call.tool
129 self.executed_args = call.arguments
130
131 # Invoke start callback if provided
132 if on_tool_start:
133 await on_tool_start(call)
134
135 if self.raise_exception:
136 now = datetime.now(UTC)
137 yield CTPToolResult(
138 id=call.id,
139 tool=call.tool,
140 result=None,
141 error="Simulated execute_tool exception",
142 start_time=now,
143 end_time=now,

Calls

no outgoing calls