From 99b940f8a75d92b47167ab5c9fb6c5a0a92dfa46 Mon Sep 17 00:00:00 2001 From: sravan27 Date: Thu, 21 May 2026 15:38:23 +0530 Subject: [PATCH 1/2] Add Omi Uber call chat tool app --- plugins/uber_call/README.md | 63 +++++++++++ plugins/uber_call/__init__.py | 2 + plugins/uber_call/main.py | 153 +++++++++++++++++++++++++++ plugins/uber_call/requirements.txt | 3 + plugins/uber_call/test_uber_links.py | 42 ++++++++ plugins/uber_call/uber_links.py | 116 ++++++++++++++++++++ 6 files changed, 379 insertions(+) create mode 100644 plugins/uber_call/README.md create mode 100644 plugins/uber_call/__init__.py create mode 100644 plugins/uber_call/main.py create mode 100644 plugins/uber_call/requirements.txt create mode 100644 plugins/uber_call/test_uber_links.py create mode 100644 plugins/uber_call/uber_links.py diff --git a/plugins/uber_call/README.md b/plugins/uber_call/README.md new file mode 100644 index 00000000000..e9ee42227d8 --- /dev/null +++ b/plugins/uber_call/README.md @@ -0,0 +1,63 @@ +# Omi Uber Call App + +An Omi Chat Tools app that turns a natural-language request such as +“call Uber to the airport” into a ready-to-open Uber deep link. + +The app deliberately uses Uber's public mobile deep-link flow instead of the +restricted Ride Request API. That keeps the integration usable without partner +credentials while still letting the user confirm pickup, destination, product, +pricing, and payment inside Uber before any ride is requested. + +## What It Does + +- Exposes an Omi Chat Tools manifest at `/.well-known/omi-tools.json` +- Provides a `call_uber` tool endpoint at `/api/call_uber` +- Accepts a destination, optional pickup/dropoff coordinates, and optional + product ID +- Returns both a mobile web link and an app scheme link for Uber +- Fails fast when Omi does not provide a usable destination + +## Running Locally + +```bash +cd plugins/uber_call +python3 -m venv .venv +. .venv/bin/activate +pip install -r requirements.txt +uvicorn main:app --reload --port 8080 +``` + +## Omi App Store Setup + +Create an External Integration app in Omi and set: + +- App Home URL: `https://` +- Chat Tools Manifest URL: `https:///.well-known/omi-tools.json` + +After install, the user can ask Omi: + +> Call Uber to 1455 Market Street, San Francisco + +Omi calls `/api/call_uber` and receives a confirmation-safe link: + +```json +{ + "result": "Uber is ready for 1455 Market Street, San Francisco. Open the link to review and confirm the ride in Uber: https://m.uber.com/ul/?action=setPickup&pickup=my_location&dropoff[formatted_address]=1455%20Market%20Street%2C%20San%20Francisco&dropoff[nickname]=1455%20Market%20Street%2C%20San%20Francisco", + "web_link": "https://m.uber.com/ul/?...", + "app_link": "uber://?..." +} +``` + +## Safety And Compliance Notes + +- The app does not book rides silently. +- The user must review and confirm the trip in Uber. +- No Uber OAuth token, API key, payment credential, or customer data is stored. +- If Omi provides geolocation, it is only used to prefill pickup coordinates. + +## Tests + +```bash +python3 -m unittest plugins/uber_call/test_uber_links.py +``` + diff --git a/plugins/uber_call/__init__.py b/plugins/uber_call/__init__.py new file mode 100644 index 00000000000..b60ad669bc8 --- /dev/null +++ b/plugins/uber_call/__init__.py @@ -0,0 +1,2 @@ +"""Uber Call Omi app.""" + diff --git a/plugins/uber_call/main.py b/plugins/uber_call/main.py new file mode 100644 index 00000000000..a34460a0b09 --- /dev/null +++ b/plugins/uber_call/main.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +from typing import Any + +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel, Field + +from uber_links import build_location, build_uber_deep_links + + +app = FastAPI(title="Omi Uber Call App") + + +class CallUberRequest(BaseModel): + uid: str | None = None + app_id: str | None = None + tool_name: str | None = None + destination: str | None = Field( + default=None, + description="The destination the user wants to go to, such as 'SFO airport'.", + ) + pickup_address: str | None = None + pickup_latitude: float | None = None + pickup_longitude: float | None = None + dropoff_address: str | None = None + dropoff_latitude: float | None = None + dropoff_longitude: float | None = None + product_id: str | None = None + geolocation: dict[str, Any] | None = None + + +def _geo_value(geolocation: dict[str, Any] | None, *keys: str) -> Any: + if not geolocation: + return None + for key in keys: + value = geolocation.get(key) + if value not in (None, ""): + return value + return None + + +@app.get("/health") +def health() -> dict[str, str]: + return {"status": "ok"} + + +@app.get("/.well-known/omi-tools.json") +def omi_tools_manifest() -> dict[str, Any]: + return { + "tools": [ + { + "name": "call_uber", + "description": ( + "Prepare an Uber ride link for the user's requested destination. " + "Use this when the user asks to call, book, or open Uber. " + "The user must confirm the ride inside Uber." + ), + "endpoint": "/api/call_uber", + "method": "POST", + "parameters": { + "properties": { + "destination": { + "type": "string", + "description": "Required destination name or address, for example 'SFO airport'.", + }, + "dropoff_address": { + "type": "string", + "description": "Optional exact dropoff formatted address.", + }, + "dropoff_latitude": { + "type": "number", + "description": "Optional dropoff latitude.", + }, + "dropoff_longitude": { + "type": "number", + "description": "Optional dropoff longitude.", + }, + "pickup_address": { + "type": "string", + "description": "Optional pickup formatted address. Defaults to user's current location.", + }, + "pickup_latitude": { + "type": "number", + "description": "Optional pickup latitude.", + }, + "pickup_longitude": { + "type": "number", + "description": "Optional pickup longitude.", + }, + "product_id": { + "type": "string", + "description": "Optional Uber product_id if the app owner wants to preselect a ride type.", + }, + }, + "required": ["destination"], + }, + "auth_required": False, + "status_message": "Preparing Uber ride link...", + } + ] + } + + +@app.post("/api/call_uber") +def call_uber(payload: CallUberRequest) -> dict[str, str]: + pickup_latitude = ( + payload.pickup_latitude + if payload.pickup_latitude is not None + else _geo_value(payload.geolocation, "latitude", "lat") + ) + pickup_longitude = ( + payload.pickup_longitude + if payload.pickup_longitude is not None + else _geo_value(payload.geolocation, "longitude", "lng", "lon") + ) + pickup_address = payload.pickup_address or _geo_value(payload.geolocation, "formatted_address", "address") + + pickup = build_location( + latitude=pickup_latitude, + longitude=pickup_longitude, + formatted_address=pickup_address, + nickname=( + "Current location" + if pickup_address or (pickup_latitude is not None and pickup_longitude is not None) + else None + ), + ) + dropoff = build_location( + latitude=payload.dropoff_latitude, + longitude=payload.dropoff_longitude, + formatted_address=payload.dropoff_address or payload.destination, + nickname=payload.destination or payload.dropoff_address, + ) + + try: + links = build_uber_deep_links( + destination=payload.destination, + pickup=pickup, + dropoff=dropoff, + product_id=payload.product_id, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + destination = payload.destination or payload.dropoff_address or "the selected destination" + return { + "result": ( + f"Uber is ready for {destination}. Open this link to review and confirm the ride in Uber: " + f"{links.web_link}" + ), + "web_link": links.web_link, + "app_link": links.app_link, + } diff --git a/plugins/uber_call/requirements.txt b/plugins/uber_call/requirements.txt new file mode 100644 index 00000000000..a3db7023aca --- /dev/null +++ b/plugins/uber_call/requirements.txt @@ -0,0 +1,3 @@ +fastapi>=0.111.0 +uvicorn[standard]>=0.30.0 + diff --git a/plugins/uber_call/test_uber_links.py b/plugins/uber_call/test_uber_links.py new file mode 100644 index 00000000000..f8c3fb40594 --- /dev/null +++ b/plugins/uber_call/test_uber_links.py @@ -0,0 +1,42 @@ +import unittest +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from uber_links import build_location, build_uber_deep_links + + +class UberDeepLinkTests(unittest.TestCase): + def test_destination_builds_mobile_web_and_app_links(self): + links = build_uber_deep_links(destination="SFO Airport") + + self.assertTrue(links.web_link.startswith("https://m.uber.com/ul/?")) + self.assertTrue(links.app_link.startswith("uber://?")) + self.assertIn("action=setPickup", links.web_link) + self.assertIn("pickup=my_location", links.web_link) + self.assertIn("dropoff[formatted_address]=SFO%20Airport", links.web_link) + self.assertIn("dropoff[nickname]=SFO%20Airport", links.web_link) + + def test_coordinates_and_product_id_are_preserved(self): + pickup = build_location(latitude=37.775818, longitude=-122.418028, formatted_address="1455 Market St") + dropoff = build_location(latitude=37.6213129, longitude=-122.3789554, formatted_address="SFO") + + links = build_uber_deep_links(pickup=pickup, dropoff=dropoff, product_id="product-123") + + self.assertIn("pickup[latitude]=37.775818", links.web_link) + self.assertIn("pickup[longitude]=-122.418028", links.web_link) + self.assertIn("dropoff[latitude]=37.6213129", links.web_link) + self.assertIn("dropoff[longitude]=-122.3789554", links.web_link) + self.assertIn("product_id=product-123", links.web_link) + + def test_destination_is_required(self): + with self.assertRaises(ValueError): + build_uber_deep_links() + + def test_whitespace_destination_is_rejected(self): + with self.assertRaises(ValueError): + build_uber_deep_links(destination=" ") + + +if __name__ == "__main__": + unittest.main() diff --git a/plugins/uber_call/uber_links.py b/plugins/uber_call/uber_links.py new file mode 100644 index 00000000000..5bf17408ffc --- /dev/null +++ b/plugins/uber_call/uber_links.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Iterable +from urllib.parse import quote + + +UBER_WEB_BASE_URL = "https://m.uber.com/ul/" +UBER_APP_BASE_URL = "uber://" + + +@dataclass(frozen=True) +class UberLocation: + latitude: float | None = None + longitude: float | None = None + nickname: str | None = None + formatted_address: str | None = None + + def has_coordinates(self) -> bool: + return self.latitude is not None and self.longitude is not None + + def has_label(self) -> bool: + return bool(self.nickname or self.formatted_address) + + +@dataclass(frozen=True) +class UberDeepLinks: + web_link: str + app_link: str + + +def _clean_text(value: str | None) -> str | None: + if value is None: + return None + cleaned = " ".join(value.split()) + return cleaned or None + + +def _normalize_float(value: float | int | str | None) -> float | None: + if value in (None, ""): + return None + return float(value) + + +def build_location( + *, + latitude: float | int | str | None = None, + longitude: float | int | str | None = None, + nickname: str | None = None, + formatted_address: str | None = None, +) -> UberLocation: + lat = _normalize_float(latitude) + lng = _normalize_float(longitude) + return UberLocation( + latitude=lat, + longitude=lng, + nickname=_clean_text(nickname), + formatted_address=_clean_text(formatted_address), + ) + + +def _append_location_params(params: list[tuple[str, str]], prefix: str, location: UberLocation) -> None: + if location.has_coordinates(): + params.append((f"{prefix}[latitude]", str(location.latitude))) + params.append((f"{prefix}[longitude]", str(location.longitude))) + if location.nickname: + params.append((f"{prefix}[nickname]", location.nickname)) + if location.formatted_address: + params.append((f"{prefix}[formatted_address]", location.formatted_address)) + + +def _encode_query(params: Iterable[tuple[str, str]]) -> str: + return "&".join(f"{quote(key, safe='[]')}={quote(value, safe='')}" for key, value in params) + + +def build_uber_deep_links( + *, + destination: str | None = None, + pickup: UberLocation | None = None, + dropoff: UberLocation | None = None, + product_id: str | None = None, +) -> UberDeepLinks: + destination_label = _clean_text(destination) + pickup_location = pickup or UberLocation() + dropoff_location = dropoff or UberLocation() + + if destination_label and not dropoff_location.has_label(): + dropoff_location = UberLocation( + latitude=dropoff_location.latitude, + longitude=dropoff_location.longitude, + nickname=destination_label, + formatted_address=destination_label, + ) + + if not destination_label and not dropoff_location.has_label() and not dropoff_location.has_coordinates(): + raise ValueError("A destination or dropoff location is required") + + params: list[tuple[str, str]] = [("action", "setPickup")] + + if pickup_location.has_coordinates() or pickup_location.has_label(): + _append_location_params(params, "pickup", pickup_location) + else: + params.append(("pickup", "my_location")) + + _append_location_params(params, "dropoff", dropoff_location) + + cleaned_product_id = _clean_text(product_id) + if cleaned_product_id: + params.append(("product_id", cleaned_product_id)) + + query = _encode_query(params) + return UberDeepLinks( + web_link=f"{UBER_WEB_BASE_URL}?{query}", + app_link=f"{UBER_APP_BASE_URL}?{query}", + ) + From 847e2c1b118ba49b57f94eb835968a1d8471f72e Mon Sep 17 00:00:00 2001 From: sravan27 Date: Thu, 21 May 2026 15:46:54 +0530 Subject: [PATCH 2/2] Fix Uber call endpoint validation --- plugins/uber_call/main.py | 34 +++++++++++------------ plugins/uber_call/requirements.txt | 2 +- plugins/uber_call/test_main.py | 43 ++++++++++++++++++++++++++++++ 3 files changed, 61 insertions(+), 18 deletions(-) create mode 100644 plugins/uber_call/test_main.py diff --git a/plugins/uber_call/main.py b/plugins/uber_call/main.py index a34460a0b09..18b5bb8dc66 100644 --- a/plugins/uber_call/main.py +++ b/plugins/uber_call/main.py @@ -58,6 +58,7 @@ def omi_tools_manifest() -> dict[str, Any]: "endpoint": "/api/call_uber", "method": "POST", "parameters": { + "type": "object", "properties": { "destination": { "type": "string", @@ -115,24 +116,23 @@ def call_uber(payload: CallUberRequest) -> dict[str, str]: ) pickup_address = payload.pickup_address or _geo_value(payload.geolocation, "formatted_address", "address") - pickup = build_location( - latitude=pickup_latitude, - longitude=pickup_longitude, - formatted_address=pickup_address, - nickname=( - "Current location" - if pickup_address or (pickup_latitude is not None and pickup_longitude is not None) - else None - ), - ) - dropoff = build_location( - latitude=payload.dropoff_latitude, - longitude=payload.dropoff_longitude, - formatted_address=payload.dropoff_address or payload.destination, - nickname=payload.destination or payload.dropoff_address, - ) - try: + pickup = build_location( + latitude=pickup_latitude, + longitude=pickup_longitude, + formatted_address=pickup_address, + nickname=( + "Current location" + if pickup_address or (pickup_latitude is not None and pickup_longitude is not None) + else None + ), + ) + dropoff = build_location( + latitude=payload.dropoff_latitude, + longitude=payload.dropoff_longitude, + formatted_address=payload.dropoff_address or payload.destination, + nickname=payload.destination or payload.dropoff_address, + ) links = build_uber_deep_links( destination=payload.destination, pickup=pickup, diff --git a/plugins/uber_call/requirements.txt b/plugins/uber_call/requirements.txt index a3db7023aca..21b54644dfc 100644 --- a/plugins/uber_call/requirements.txt +++ b/plugins/uber_call/requirements.txt @@ -1,3 +1,3 @@ fastapi>=0.111.0 +httpx>=0.27.0 uvicorn[standard]>=0.30.0 - diff --git a/plugins/uber_call/test_main.py b/plugins/uber_call/test_main.py new file mode 100644 index 00000000000..cf48c618098 --- /dev/null +++ b/plugins/uber_call/test_main.py @@ -0,0 +1,43 @@ +import sys +import unittest +from pathlib import Path + +try: + from fastapi.testclient import TestClient +except ModuleNotFoundError: + TestClient = None + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +if TestClient is not None: + from main import app + + +@unittest.skipIf(TestClient is None, "fastapi/httpx test dependencies are not installed") +class CallUberEndpointTests(unittest.TestCase): + def setUp(self): + self.client = TestClient(app) + + def test_manifest_parameters_are_json_schema_object(self): + response = self.client.get("/.well-known/omi-tools.json") + + self.assertEqual(response.status_code, 200) + parameters = response.json()["tools"][0]["parameters"] + self.assertEqual(parameters["type"], "object") + self.assertIn("destination", parameters["required"]) + + def test_bad_geolocation_returns_400(self): + response = self.client.post( + "/api/call_uber", + json={ + "destination": "SFO Airport", + "geolocation": {"latitude": "nearby", "longitude": "-122.418028"}, + }, + ) + + self.assertEqual(response.status_code, 400) + self.assertIn("could not convert string to float", response.json()["detail"]) + + +if __name__ == "__main__": + unittest.main()