update cli with typer

Author: grll

pyproject.toml

@@ -5,8 +5,8 @@ description = "An open source proxy for MCP servers"
 readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
-    "logfire>=2.6.2",
-    "mcp>=1.1.0",
+    "mcp>=1.2.0rc1",
+    "typer>=0.15.1",
 ]
 
 [tool.hatch.version]
@@ -31,9 +31,6 @@ addopts = [
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
 
-[tool.uv.sources]
-mcp = { git = "https://github.com/modelcontextprotocol/python-sdk.git", rev = "main" }
-
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

src/omproxy/cli.py

@@ -1,90 +1,83 @@
 #!/usr/bin/env python3
 
-import argparse
-import logging
 import os
-import uuid
-from contextvars import ContextVar
-from pathlib import Path
+from typing import Optional
+import typer
+from typing_extensions import Annotated
 
-import anyio
-import logfire
 from mcp.client.stdio import StdioServerParameters
-
 from omproxy import __version__
-from omproxy.highlevel_proxy import run_stdio_client
-
-# Create a global context variable for instance_id
-instance_id_var = ContextVar("instance_id", default=None)
-
-
-def get_or_create_instance_id() -> str:
-    """Get or create a persistent UUID for this proxy instance."""
-    id_file = Path.home() / ".omproxy" / "instance_id"
-    id_file.parent.mkdir(parents=True, exist_ok=True)
-
-    if id_file.exists():
-        return id_file.read_text().strip()
-
-    instance_id = str(uuid.uuid4())
-    id_file.write_text(instance_id)
-    return instance_id
-
-
-def main():
-    parser = argparse.ArgumentParser(
-        description="Bidirectional proxy for subprocess communication"
-    )
-    parser.add_argument(
-        "--name",
-        "-n",
-        type=str,
-        help="Name of the service",
-    )
-    parser.add_argument(
-        "--version", action="version", version=__version__, help="Show version and exit"
-    )
-    parser.add_argument(
-        "-v", "--verbose", action="store_true", help="Enable debug logging"
-    )
-    parser.add_argument("command", help="Command to run with optional arguments")
-    parser.add_argument(
-        "args", nargs=argparse.REMAINDER, help="Arguments to pass to the command"
-    )
-    args = parser.parse_args()
-
-    # TODO: (use auth see https://github.com/pydantic/logfire/issues/651#issuecomment-2522714987)
-    os.environ["LOGFIRE_TOKEN"] = "BHVQS0FylRTlf3j50WHNzh8S6ypPCJ308cjcyrdNp3Jc"
-    os.environ["LOGFIRE_PROJECT_NAME"] = "iod-mcp"
-    os.environ["LOGFIRE_PROJECT_URL"] = "https://logfire.pydantic.dev/grll/iod-mcp"
-    os.environ["LOGFIRE_API_URL"] = "https://logfire-api.pydantic.dev"
-
-    instance_id = get_or_create_instance_id()
-    instance_id_var.set(instance_id)
-
-    # Configure logging
-    logfire.configure(
-        service_name=f"omproxy[{args.name}]",
-        service_version=__version__,
-        console=False,
+from omproxy.proxy import SseProxy, StdioProxy
+
+app = typer.Typer(no_args_is_help=True)
+
+
+def version_callback(value: bool):
+    if value:
+        typer.echo(f"omproxy version: {__version__}")
+        raise typer.Exit()
+
+
+@app.callback()
+def callback(
+    version: Annotated[
+        bool,
+        typer.Option(
+            "--version", callback=version_callback, help="Show version and exit"
+        ),
+    ] = False,
+):
+    """Bidirectional proxy for subprocess communication."""
+    pass
+
+
+@app.command()
+def sse(
+    url: Annotated[str, typer.Option(help="SSE server URL")],
+    headers: Annotated[
+        Optional[str], typer.Option(help="SSE headers as key1=value1,key2=value2")
+    ] = None,
+    timeout: Annotated[float, typer.Option(help="SSE connection timeout")] = 5.0,
+    sse_read_timeout: Annotated[float, typer.Option(help="SSE read timeout")] = 300.0,
+):
+    """Use SSE proxy protocol"""
+    # Parse headers if provided
+    headers_dict = {}
+    if headers:
+        try:
+            headers_dict = dict(h.split("=") for h in headers.split(","))
+        except ValueError:
+            typer.echo(
+                "Error: Invalid headers format. Use key1=value1,key2=value2", err=True
+            )
+            raise typer.Exit(1)
+
+    proxy = SseProxy()
+    proxy.run(
+        url=url,
+        headers=headers_dict,
+        timeout=timeout,
+        sse_read_timeout=sse_read_timeout,
     )
-    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
 
-    logfire.info(
-        "starting_proxy", command=args.command, args=args.args, instance_id=instance_id
-    )
 
-    async def run_proxy():
-        await run_stdio_client(
-            StdioServerParameters(
-                command=args.command,
-                args=args.args,
-                env=os.environ,
-            )
+@app.command()
+def stdio(
+    command: Annotated[str, typer.Argument(help="Command to run")],
+    args: Annotated[
+        Optional[list[str]], typer.Argument(help="Arguments for the command")
+    ] = None,
+):
+    """Use stdio proxy protocol"""
+    proxy = StdioProxy()
+    proxy.run(
+        StdioServerParameters(
+            command=command,
+            args=args or [],
+            env=os.environ,
         )
-
-    anyio.run(run_proxy)
+    )
 
 
 if __name__ == "__main__":
-    main()
+    app()