Add uvx client

This commit is contained in:
Jeremiah Lowin 2025-04-07 11:23:14 -04:00
commit 6440c73846

View file

@ -1,4 +1,6 @@
import contextlib
import os
from typing import Dict, List, Optional
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
@ -39,3 +41,95 @@ class StdioClient(BaseClient):
) as session:
async with self._set_session(transport, session):
yield self
class UvxClient(BaseClient):
"""Client that uses uvx to run Python tools in isolated environments.
uvx automatically installs and manages dependencies from pyproject.toml.
"""
def __init__(
self,
tool_name: str,
tool_args: Optional[List[str]] = None,
project_directory: Optional[str] = None,
python_version: Optional[str] = None,
with_packages: Optional[List[str]] = None,
from_package: Optional[str] = None,
env_vars: Optional[Dict[str, str]] = None,
**kwargs: Unpack[ClientKwargs],
):
"""Initialize a UvxClient that uses uvx to run Python tools in isolated environments.
Args:
tool_name: Name of the tool/command to run
tool_args: Arguments to pass to the tool
project_directory: Path to the project directory (optional)
python_version: Specific Python version to use (e.g., "3.10")
with_packages: Additional packages to include
from_package: Package that provides the tool if different from tool_name
env_vars: Environment variables to set for the process
**kwargs: Additional arguments for BaseClient
"""
super().__init__(**kwargs)
self.tool_name = tool_name
self.tool_args = tool_args or []
self.project_directory = project_directory
self.python_version = python_version
self.with_packages = with_packages or []
self.from_package = from_package
self.env_vars = env_vars or {}
@contextlib.asynccontextmanager
async def _connect(self):
"""Set up uvx connection and session"""
# Check if project directory exists if provided
if self.project_directory and not os.path.isdir(self.project_directory):
raise ValueError(
f"Project directory does not exist: {self.project_directory}"
)
# Build the uvx command arguments
args = []
# Add Python version if specified
if self.python_version:
args.extend(["--python", self.python_version])
# Add from package if specified
if self.from_package:
args.extend(["--from", self.from_package])
# Add with packages if specified
for pkg in self.with_packages:
args.extend(["--with", pkg])
# Add the tool name
args.append(self.tool_name)
# Add the tool arguments
args.extend(self.tool_args)
# Create environment variables dictionary
env = os.environ.copy()
env.update(self.env_vars)
# Configure the server parameters
server_params = StdioServerParameters(
command="uvx",
args=args,
env=env,
cwd=self.project_directory,
)
async with stdio_client(server_params) as transport:
stdio, write = transport
async with ClientSession(
read_stream=stdio,
write_stream=write,
**self._session_kwargs(),
) as session:
async with self._set_session(transport, session):
yield self