use http.client instead of urllib3

libs/cli/langchain_cli/cli.py

@@ -6,7 +6,7 @@
 from langchain_cli.namespaces import app as app_namespace
 from langchain_cli.namespaces import template as template_namespace
 
-__version__ = "0.0.10"
+__version__ = "0.0.11"
 
 app = typer.Typer(no_args_is_help=True, add_completion=False)
 app.add_typer(

libs/cli/langchain_cli/utils/events.py

@@ -1,8 +1,7 @@
+import http.client
 import json
 from typing import Any, Dict, List, Optional, TypedDict
 
-import urllib3
-
 WRITE_KEY = "310apTK0HUFl4AOv"
 
 
@@ -11,43 +10,32 @@ class EventDict(TypedDict):
     properties: Optional[Dict[str, Any]]
 
 
-def create_event(event: EventDict) -> None:
-    """
-    Creates a new event with the given type and payload.
-    """
-    data = {
-        "write_key": WRITE_KEY,
-        "event": event["event"],
-        "properties": event.get("properties"),
-    }
-    try:
-        urllib3.request(
-            "POST",
-            "https://app.firstpartyhq.com/events/v1/track",
-            body=json.dumps(data),
-            headers={"Content-Type": "application/json"},
-        )
-    except Exception:
-        pass
-
-
-def create_events(events: List[EventDict]) -> None:
-    data = {
-        "events": [
-            {
-                "write_key": WRITE_KEY,
-                "event": event["event"],
-                "properties": event.get("properties"),
-            }
-            for event in events
-        ]
-    }
+def create_events(events: List[EventDict]) -> Optional[Any]:
     try:
-        urllib3.request(
-            "POST",
-            "https://app.firstpartyhq.com/events/v1/track/bulk",
-            body=json.dumps(data),
-            headers={"Content-Type": "application/json"},
-        )
+        data = {
+            "events": [
+                {
+                    "write_key": WRITE_KEY,
+                    "name": event["event"],
+                    "properties": event.get("properties"),
+                }
+                for event in events
+            ]
+        }
+
+        conn = http.client.HTTPSConnection("app.firstpartyhq.com")
+
+        payload = json.dumps(data)
+
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        }
+
+        conn.request("POST", "/events/v1/track/bulk", payload, headers)
+
+        res = conn.getresponse()
+
+        return json.loads(res.read())
     except Exception:
-        pass
+        return None

libs/cli/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langchain-cli"
-version = "0.0.10"
+version = "0.0.11"
 description = "CLI for interacting with LangChain"
 authors = ["Erick Friis <erick@langchain.dev>"]
 readme = "README.md"

libs/cli/tests/test_events.py

@@ -0,0 +1,7 @@
+from langchain_cli.utils.events import EventDict, create_events
+
+
+def test_create_events() -> None:
+    assert create_events(
+        [EventDict(event="Test Event", properties={"test": "test"})]
+    ) == {"status": "success"}