mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-09-10 22:41:06 +00:00
041f957a7f
* refactor(components): extract shared component state into mixin - Introduce ComponentMixin class with shared state for components and steps - Move identity, config, and vault path functionality to ComponentMixin - Update BaseComponent to inherit from ComponentMixin - Update BaseStep to inherit from ComponentMixin - Consolidate vault path helper methods in ComponentMixin - Remove duplicate vault path implementations from BaseComponent and BaseStep - Add ComponentMixin to components module exports * refactor(file_io): implement path locks cache eviction mechanism - Add _PATH_LOCKS_MAX constant set to 1024 for cache size limit - Implement cache eviction logic when locks exceed maximum capacity - Remove half of unlocked entries when cache limit is reached - Use list comprehension to identify unlocked locks for removal - Maintain existing path normalization and locking behavior fix(edit): correct method call from public to private fail method - Change self.fail to self._fail for internal error handling - Maintain consistent private method usage within class fix(mcp_client): change pop to get for optional command and args - Replace kwargs.pop with kwargs.get to avoid removing keys - Preserve original kwargs dictionary contents - Maintain default empty string and list values feat(reme): add client backend validation with error raising - Check if client_cls is None before instantiation - Raise ValueError with descriptive message for unknown backends - Provide clear error feedback for invalid backend configurations * fix(components): move directory creation to start method - Moved component_metadata_path.mkdir call from __init__ to _start in base_keyword_index - Moved component_metadata_path.mkdir call from __init__ to _start in local_file_graph - Moved component_metadata_path.mkdir call from __init__ to _start in local_file_store - Ensures directory creation happens after component initialization - Prevents potential issues with path creation during object construction * fix(steps): replace assertions with runtime errors for app_context validation - Replace assert statements with explicit RuntimeError exceptions when app_context is None - Add descriptive error messages for better debugging when resolving components - Replace assert in resolve_component method with proper exception handling - Replace assert in get_file_parser method with proper exception handling - Maintain same functionality while improving error reporting clarity * refactor(file_io): split file IO utilities into modular components - Move daily note helpers to separate _daily_index module - Extract path validation and resolution to new _path module - Remove unused code and imports from _file_io module - Update import statements across affected modules - Introduce WikilinkHandler utility for link parsing - Replace regex-based link extraction with WikilinkHandler - Add integration JSONL files to gitignore - Consolidate file locking mechanism in _file_io module * style(formatter): fix spacing issues in file IO and chunked file parser - Fixed whitespace around colon in slice notation in file_io.py - Corrected spacing around colon in slice notation in chunked_file_parser.py - Applied consistent formatting for array slicing operations - Improved code readability by standardizing space placement in ranges * refactor(steps): replace property-based component resolution with Ref descriptor - Introduce Ref descriptor class for lazy component dependency resolution - Replace _resolve method and individual properties with Ref descriptors - Add as_llm, as_llm_formatter, as_token_counter, file_store, and embedding Ref attributes - Remove legacy property methods and resolve logic from BaseStep - Add cache clearing mechanism for Ref values during step calls - Update UpdateCatalogStep to use Ref instead of property-based resolution
123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
"""MCP client for ReMe services."""
|
|
|
|
import json
|
|
import os
|
|
from collections.abc import AsyncGenerator
|
|
from typing import Any
|
|
|
|
from fastmcp import Client
|
|
from fastmcp.client import SSETransport, StdioTransport, StreamableHttpTransport
|
|
from fastmcp.client.client import CallToolResult
|
|
|
|
from .base_client import BaseClient
|
|
from ..component_registry import R
|
|
from ...constants import REME_SERVICE_INFO, REME_DEFAULT_HOST, REME_DEFAULT_PORT
|
|
|
|
_TRANSPORT_MAP = {
|
|
"sse": SSETransport,
|
|
"stdio": StdioTransport,
|
|
"streamable-http": StreamableHttpTransport,
|
|
}
|
|
|
|
|
|
@R.register("mcp")
|
|
class MCPClient(BaseClient):
|
|
"""MCP client that communicates with ReMe MCP service via fastmcp.Client.
|
|
|
|
Usage:
|
|
# SSE (default)
|
|
client = MCPClient(host="localhost", port=8000)
|
|
async with client:
|
|
async for text in client(action="my_tool", query="hello"):
|
|
print(text)
|
|
|
|
# Streamable HTTP
|
|
client = MCPClient(transport="streamable-http", host="localhost", port=8000)
|
|
|
|
# Stdio
|
|
client = MCPClient(transport="stdio", command="python", args=["server.py"])
|
|
|
|
# Custom transport object
|
|
from fastmcp.client import SSETransport
|
|
client = MCPClient(transport=SSETransport(url="http://host:port/sse"))
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
transport: str | Any = "sse",
|
|
host: str | None = None,
|
|
port: int | None = None,
|
|
timeout: float = 30.0,
|
|
**kwargs,
|
|
):
|
|
super().__init__(**kwargs)
|
|
|
|
if isinstance(transport, str) and transport not in _TRANSPORT_MAP:
|
|
raise ValueError(f"Unknown transport: {transport!r}, expected one of {list(_TRANSPORT_MAP)}")
|
|
|
|
if isinstance(transport, str) and transport != "stdio":
|
|
if not (host and port):
|
|
if service_info := os.environ.get(REME_SERVICE_INFO):
|
|
try:
|
|
data = json.loads(service_info)
|
|
host = data["host"]
|
|
port = data["port"]
|
|
except Exception:
|
|
self.logger.warning(f"Invalid service info: {service_info}")
|
|
host, port = REME_DEFAULT_HOST, REME_DEFAULT_PORT
|
|
else:
|
|
host, port = REME_DEFAULT_HOST, REME_DEFAULT_PORT
|
|
self.host = host
|
|
self.port = port
|
|
|
|
self.transport = transport
|
|
self.timeout = timeout
|
|
|
|
def _build_transport(self):
|
|
if not isinstance(self.transport, str):
|
|
return self.transport
|
|
|
|
cls = _TRANSPORT_MAP[self.transport]
|
|
|
|
if self.transport == "stdio":
|
|
command = self.kwargs.get("command", "")
|
|
args = self.kwargs.get("args", [])
|
|
return cls(command=command, args=args)
|
|
|
|
path = "/sse" if self.transport == "sse" else "/mcp"
|
|
url = f"http://{self.host}:{self.port}{path}"
|
|
return cls(url=url)
|
|
|
|
# pylint: disable=unnecessary-dunder-call
|
|
async def _start(self) -> None:
|
|
if self.client is None:
|
|
self.client = Client(self._build_transport(), timeout=self.timeout)
|
|
await self.client.__aenter__()
|
|
|
|
# pylint: disable=invalid-overridden-method
|
|
async def _execute(self, action: str, payload: dict) -> AsyncGenerator[str, None]:
|
|
if self.client is None:
|
|
raise RuntimeError("Client not initialized. Call _start() first.")
|
|
|
|
result: CallToolResult = await self.client.call_tool(action, payload)
|
|
yield self._extract_text(result)
|
|
|
|
async def list_actions(self) -> list[dict]:
|
|
"""Return raw MCP Tool dumps; each dict gets an `action` key (the tool name)."""
|
|
if self.client is None:
|
|
raise RuntimeError("Client not initialized. Call _start() first.")
|
|
tools = await self.client.list_tools()
|
|
return [tool.model_dump() for tool in tools]
|
|
|
|
# pylint: disable=unnecessary-dunder-call
|
|
async def _close(self) -> None:
|
|
if self.client is not None:
|
|
await self.client.__aexit__(None, None, None)
|
|
self.client = None
|
|
|
|
@staticmethod
|
|
def _extract_text(result: CallToolResult) -> str:
|
|
for block in result.content:
|
|
if hasattr(block, "text"):
|
|
return block.text
|
|
return str(result.content)
|