diff --git a/entrypoint.sh b/entrypoint.sh index ddd01afe..23dd5bb3 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -24,14 +24,14 @@ SAFE_VARS="HOME USER PATH HOSTNAME TERM LANG LC_ALL \ GATEWAY_URL PYTHONDONTWRITEBYTECODE PYTHONUNBUFFERED \ HF_HOME SENTENCE_TRANSFORMERS_HOME HF_HUB_OFFLINE TRANSFORMERS_OFFLINE \ OMEGACLAW_DIR MEMORY_DIR LLM_SERVER_LOCAL_URL TEST_SERVER_IP \ - WS_URL WS_TOKEN" + WS_URL WS_TOKEN MCP_JSON_CONTENT" -env_args="" +env_args=() for var in $SAFE_VARS; do - eval val=\${$var:-} + val="${!var:-}" if [ -n "$val" ]; then - env_args="$env_args $var=$val" + env_args+=("$var=$val") fi done -exec env -i $env_args su nobody -s /bin/sh -c "sh run.sh run.metta $*" +exec env -i "${env_args[@]}" su nobody -s /bin/sh -c "sh run.sh run.metta $*" diff --git a/lib_omegaclaw.metta b/lib_omegaclaw.metta index 633db54b..250e796a 100644 --- a/lib_omegaclaw.metta +++ b/lib_omegaclaw.metta @@ -7,6 +7,7 @@ !(import! &self (library OmegaClaw-Core lib_llm_ext.py)) !(import! &self (library OmegaClaw-Core ./src/helper.py)) !(import! &self (library OmegaClaw-Core ./src/agentverse.py)) +!(import! &self (library OmegaClaw-Core ./src/mcp_client.py)) !(import! &self (library OmegaClaw-Core ./channels/wschat.py)) !(import! &self (library OmegaClaw-Core ./channels/irc.py)) !(import! &self (library OmegaClaw-Core ./channels/mattermost.py)) diff --git a/requirements.txt b/requirements.txt index 01352a5a..f020334b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,4 +9,5 @@ import-kb==0.1.8 py-landlock==0.1.1 pyyaml==6.0.3 ddgs==9.14.4 -websockets +websockets==16.0 +mcp==1.27.2 diff --git a/scripts/omegaclaw b/scripts/omegaclaw index 0a7b3563..a1e7ed30 100755 --- a/scripts/omegaclaw +++ b/scripts/omegaclaw @@ -568,6 +568,7 @@ start() { -e OMEGACLAW_AUTH_SECRET="$OMEGACLAW_AUTH_SECRET" -e LLM_SERVER_LOCAL_URL="$llm_server_local_url" -e IMPORT_KB_ON_START="${IMPORT_KB_ON_START:-1}" + -e "MCP_JSON_CONTENT=${MCP_JSON_CONTENT:-}" "$image" "commchannel=${commchannel}" "provider=${provider}" diff --git a/src/helper.py b/src/helper.py index 31626303..7e254357 100644 --- a/src/helper.py +++ b/src/helper.py @@ -18,6 +18,8 @@ "tavily-search", "technical-analysis", "write-file", + "get-mcp-tools", + "call-mcp" } @@ -135,7 +137,7 @@ def _merge_send_continuations(lines): def balance_parentheses(s): s = s.replace("_quote_", '"').replace("_newline_", "\n") sexprs = [] - special_two_arg_cmds = {"write-file", "append-file"} + special_two_arg_cmds = {"write-file", "append-file", "call-mcp"} lines = [line.strip() for line in s.splitlines() if line.strip()] lines = _merge_send_continuations(lines) for line in lines: @@ -219,6 +221,7 @@ def test_balance_parenthesis(): assert balance_parentheses('write-file "test.txt" "hello world"') == '((write-file "test.txt" "hello world"))' assert balance_parentheses('write-file test.txt "hello world"') == '((write-file "test.txt" "hello world"))' assert balance_parentheses('send test.xt hello world') == '((send "test.xt hello world"))' + assert balance_parentheses('call-mcp get_user_agents {}') == '((call-mcp "get_user_agents" "{}"))' assert balance_parentheses('send Here are the planets:\n1. Mercury\n2. Venus') == '((send "Here are the planets:\\n1. Mercury\\n2. Venus"))' assert balance_parentheses('send Here are the options:\n- MacBook Air\n- ThinkPad X1\npin done') == '((send "Here are the options:\\n- MacBook Air\\n- ThinkPad X1") (pin "done"))' assert balance_parentheses('send "Plain text version:"\n**Mars** - red planet\nNote: Pluto is a dwarf planet') == '((send "Plain text version:\\n**Mars** - red planet\\nNote: Pluto is a dwarf planet"))' diff --git a/src/mcp_client.py b/src/mcp_client.py new file mode 100644 index 00000000..1dad711d --- /dev/null +++ b/src/mcp_client.py @@ -0,0 +1,253 @@ +import asyncio +import json +import logging +import os +import time +from contextlib import asynccontextmanager +from pathlib import Path + +import httpx +from mcp import ClientSession +from mcp.client.sse import sse_client +from mcp.client.streamable_http import streamable_http_client + +CONFIG_PATH = Path(__file__).parents[1].joinpath("mcp.json") +MCP_JSON_CONTENT = os.environ.get("MCP_JSON_CONTENT") + +SERVERS_CONFIG_MAP = {} +TOOL_ROUTING_MAP = {} # tool name -> server name +LAST_TOOL_LIST = [] +LAST_REFRESH_TIME = 0 +CACHE_TTL_SECONDS = 300 +MCP_OPERATION_TIMEOUT_SECONDS = 30 + + +def _get_logger(): + _logger = logging.getLogger("MCPClientLogger") + _logger.setLevel(logging.DEBUG) + _logger.propagate = False + + if _logger.handlers: + return _logger + + stream_handler = logging.StreamHandler() + stream_handler.setLevel(logging.DEBUG) + + log_format = logging.Formatter( + "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + ) + stream_handler.setFormatter(log_format) + + _logger.addHandler(stream_handler) + + return _logger + + +logger = _get_logger() +logger.debug(f"MCP_JSON_CONTENT configured: {bool(MCP_JSON_CONTENT)}") + + +def _run_async(coro): + try: + return asyncio.run(coro) + except RuntimeError as e: + logger.error(f"Error during _run_async: {str(e)}") + loop = asyncio.get_event_loop() + if loop.is_running(): + return asyncio.ensure_future(coro, loop=loop) + return loop.run_until_complete(coro) + + +def _load_mcp_config_to_memory(): + global SERVERS_CONFIG_MAP, MCP_JSON_CONTENT + + if MCP_JSON_CONTENT and MCP_JSON_CONTENT.strip(): + try: + config = json.loads(MCP_JSON_CONTENT) + SERVERS_CONFIG_MAP = config.get("mcpServers", {}) + logger.info(f"Configured MCP servers: {list(SERVERS_CONFIG_MAP)}") + return + except json.JSONDecodeError as e: + logger.error(f"Failed to parse inner MCP_JSON_CONTENT string atom: {str(e)}") + + if not os.path.exists(CONFIG_PATH): + SERVERS_CONFIG_MAP = {} + logger.info("Configured MCP servers: []") + return + + with open(CONFIG_PATH, "r") as f: + config = json.load(f) + + SERVERS_CONFIG_MAP = config.get("mcpServers", {}) + logger.info(f"Configured MCP servers: {list(SERVERS_CONFIG_MAP)}") + + +@asynccontextmanager +async def _connect_to_server(server_name: str, cfg: dict): + transport = cfg.get("transport", "sse") + headers = cfg.get("headers", {}) + + if transport == "sse": + async with sse_client(url=cfg["url"], headers=headers) as streams: + yield streams + return + + if transport == "streamable-http": + async with httpx.AsyncClient( + headers=headers, + follow_redirects=True, + timeout=httpx.Timeout(MCP_OPERATION_TIMEOUT_SECONDS), + ) as http_client: + async with streamable_http_client( + url=cfg["url"], + http_client=http_client, + ) as (read_stream, write_stream, _): + yield read_stream, write_stream + return + + raise ValueError( + f"Unsupported MCP transport '{transport}' for server '{server_name}'" + ) + + +async def _discover_and_map_server(server_name: str, cfg: dict) -> list[str]: + global TOOL_ROUTING_MAP + + if "url" not in cfg: + return [] + + try: + async with asyncio.timeout(MCP_OPERATION_TIMEOUT_SECONDS): + async with _connect_to_server(server_name, cfg) as (r, w): + async with ClientSession(r, w) as session: + await session.initialize() + res = await session.list_tools() + logger.info(f"List tools for server {server_name}: {res}") + # return res.model_dump_json() + + # tools_found = [] + # for tool in res.tools: + # TOOL_ROUTING_MAP[tool.name] = server_name + # + # schema = tool.inputSchema.get("properties", {}) + # tools_found.append( + # (tool.name, tool.description, list(schema.keys())) + # ) + # return tools_found + + tools_found = [] + for tool in res.tools: + TOOL_ROUTING_MAP[tool.name] = server_name + tools_found.append(tool.model_dump_json()) + logger.info(f"TOOL_ROUTING_MAP: {TOOL_ROUTING_MAP}") + return tools_found + + except Exception as e: + logger.error(f"Failed to scan tools from server '{server_name}': {str(e)}") + return [] + + +# we probably should use persistent session implementation to reuse session for multiple tool calls +# and not to reconnect every time we call mcp server +async def _execute_tool_on_server( + server_name: str, cfg: dict, tool_name: str, arguments: dict +): + async with asyncio.timeout(MCP_OPERATION_TIMEOUT_SECONDS): + async with _connect_to_server(server_name, cfg) as (r, w): + async with ClientSession(r, w) as session: + await session.initialize() + logger.info(f"Calling {tool_name} tool with arguments: {arguments}") + return await session.call_tool(tool_name, arguments=arguments) + + +def _update_server_tools_if_needed(force_update: bool = False): + logger.info("Updating server tools") + global \ + TOOL_ROUTING_MAP, \ + LAST_TOOL_LIST, \ + LAST_REFRESH_TIME, \ + SERVERS_CONFIG_MAP + + if not SERVERS_CONFIG_MAP: + _load_mcp_config_to_memory() + + current_time = time.time() + cache_is_expired = (current_time - LAST_REFRESH_TIME) > CACHE_TTL_SECONDS + + if LAST_TOOL_LIST and not cache_is_expired and not force_update: + logger.info("No need to update") + return + + TOOL_ROUTING_MAP.clear() + + all_tasks = [ + _discover_and_map_server(name, cfg) for name, cfg in SERVERS_CONFIG_MAP.items() + ] + + async def _gather_tasks(): + return await asyncio.gather(*all_tasks) + + resolved_lists = _run_async(_gather_tasks()) + + formatted_skills = [] + for server_tools in resolved_lists: + # for name, desc, param_keys in tool_list: + # formatted_skills.append(f'"- {desc}: call-mcp {name} {param_keys}"') + for server_tool_as_json_string in server_tools: + formatted_skills.append(server_tool_as_json_string) + + skills_for_log = "\n\t".join(formatted_skills) + logger.info(f"MCP_TOOLS_LIST:\n{skills_for_log}") + + LAST_TOOL_LIST = formatted_skills + + LAST_REFRESH_TIME = time.time() + + +def get_tools_as_list() -> list[str]: + + _update_server_tools_if_needed() + return LAST_TOOL_LIST + + +def call_tool(name: str, parameters_input: str | dict | None = None) -> str: + global TOOL_ROUTING_MAP, SERVERS_CONFIG_MAP + + logger.debug(f"tool_name='{name}', parameters type='{type(parameters_input)}', parameters='{parameters_input}'") + + _update_server_tools_if_needed(name not in TOOL_ROUTING_MAP) + + server_name = TOOL_ROUTING_MAP.get(name) + if not server_name: + logger.error(f"Error: Tool '{name}' cannot be resolved.") + return f"Error: Tool '{name}' cannot be resolved." + + target_config = SERVERS_CONFIG_MAP.get(server_name) + if not target_config: + logger.error(f"Error: Configuration for server '{server_name}' missing.") + return f"Error: Configuration for server '{server_name}' missing." + + if parameters_input is None: + args = {} + elif isinstance(parameters_input, str): + try: + args = json.loads(parameters_input) + except json.JSONDecodeError: + args = {} + else: + args = dict(parameters_input) + + try: + tool_result = _run_async( + _execute_tool_on_server(server_name, target_config, name, args) + ) + logger.debug(f"tool_result: {tool_result}") + text_responses = [c.text for c in tool_result.content if hasattr(c, "text")] + logger.debug(f"text_responses: {text_responses}") + return "\n".join(text_responses) + except Exception as e: + if "not found" in str(e).lower() or "404" in str(e): + logger.exception(f"Error 404 Not Found. Removing tool {name} from the map.") + TOOL_ROUTING_MAP.pop(name, None) + logger.error(f"Execution Error on tool '{name}': {str(e)}") + return f"Execution Error on tool '{name}': {str(e)}" diff --git a/src/skills.metta b/src/skills.metta index 7f751b97..6544baa6 100644 --- a/src/skills.metta +++ b/src/skills.metta @@ -26,8 +26,10 @@ "You can also use PLN:" "metta (|~ ((Implication (Inheritance $1 (IntSet Feathered))" " (Inheritance $1 Bird)) (stv 1.0 0.9))" - " ((Inheritance Pingu (IntSet Feathered)) (stv 1.0 0.9)))")) - + " ((Inheritance Pingu (IntSet Feathered)) (stv 1.0 0.9)))" + ;MCP: + "- Get all MCP tools with full description and data: get-mcp-tools" + "- Call MCP tool with parameters as json string: call-mcp name parameters_as_json_string")) (= (read-file $file) (progn (translatePredicate (exists_file $file)) @@ -54,6 +56,12 @@ (= (technical-analysis $ticker) (py-call (agentverse.technical_analysis $ticker))) +(= (get-mcp-tools) + (py-call (mcp_client.get_tools_as_list))) + +(= (call-mcp $name $parameters_as_json_string) + (py-call (mcp_client.call_tool $name $parameters_as_json_string))) + !(import_prolog_functions_from_file (library OmegaClaw-Core ./src/skills.pl) (shell first_char gc read_file_tail)) (= (metta $str)