Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions plugins/uber_call/README.md
Original file line number Diff line number Diff line change
@@ -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://<your-host>`
- Chat Tools Manifest URL: `https://<your-host>/.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
```

2 changes: 2 additions & 0 deletions plugins/uber_call/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
"""Uber Call Omi app."""

153 changes: 153 additions & 0 deletions plugins/uber_call/main.py
Original file line number Diff line number Diff line change
@@ -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": {
"type": "object",
"properties": {
Comment on lines +60 to +62
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 The parameters object in the manifest is missing "type": "object", which is required by the JSON Schema specification. Some Omi tool parsers and validators that strictly check the schema structure may reject or mishandle the manifest.

Suggested change
"parameters": {
"properties": {
"parameters": {
"type": "object",
"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")

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,
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,
}
3 changes: 3 additions & 0 deletions plugins/uber_call/requirements.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
fastapi>=0.111.0
httpx>=0.27.0
uvicorn[standard]>=0.30.0
Comment on lines +1 to +3
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 httpx is not listed as a dependency but is required by fastapi.testclient.TestClient (used in the PR's own validation script). Without it, from fastapi.testclient import TestClient will raise an ImportError in any environment that installs only what is in requirements.txt.

Suggested change
fastapi>=0.111.0
uvicorn[standard]>=0.30.0
fastapi>=0.111.0
httpx>=0.27.0
uvicorn[standard]>=0.30.0

43 changes: 43 additions & 0 deletions plugins/uber_call/test_main.py
Original file line number Diff line number Diff line change
@@ -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()
42 changes: 42 additions & 0 deletions plugins/uber_call/test_uber_links.py
Original file line number Diff line number Diff line change
@@ -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()
Loading