MCP2Lambda
Official
by danilop
- mcp_client_bedrock
from typing import Any, Dict, List, Callable
class ConverseToolManager:
def __init__(self):
self._tools = {}
self._name_mapping = {} # Maps sanitized names to original names
def _sanitize_name(self, name: str) -> str:
"""Convert hyphenated names to underscore format"""
return name.replace('-', '_')
def register_tool(self, name: str, func: Callable, description: str, input_schema: Dict):
"""
Register a new tool with the system, sanitizing the name for Bedrock compatibility
"""
sanitized_name = self._sanitize_name(name)
self._name_mapping[sanitized_name] = name
self._tools[sanitized_name] = {
'function': func,
'description': description,
'input_schema': input_schema,
'original_name': name
}
def get_tools(self) -> Dict[str, List[Dict]]:
"""
Generate the tools specification using sanitized names
"""
tool_specs = []
for sanitized_name, tool in self._tools.items():
tool_specs.append({
'toolSpec': {
'name': sanitized_name, # Use sanitized name for Bedrock
'description': tool['description'],
'inputSchema': tool['input_schema']
}
})
return {'tools': tool_specs}
async def execute_tool(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Execute a tool based on the agent's request, handling name translation
"""
tool_use_id = payload['toolUseId']
sanitized_name = payload['name']
tool_input = payload['input']
if sanitized_name not in self._tools:
raise ValueError(f"Unknown tool: {sanitized_name}")
try:
tool_func = self._tools[sanitized_name]['function']
# Use original name when calling the actual function
original_name = self._tools[sanitized_name]['original_name']
result = await tool_func(original_name, tool_input)
return {
'toolUseId': tool_use_id,
'content': [{
'text': str(result)
}],
'status': 'success'
}
except Exception as e:
return {
'toolUseId': tool_use_id,
'content': [{
'text': f"Error executing tool: {str(e)}"
}],
'status': 'error'
}
def clear_tools(self):
"""Clear all registered tools"""
self._tools.clear()