from __future__ import annotations import uuid from dataclasses import dataclass from typing import Any, Awaitable, Callable from traderai.memory import MemoryStore from traderai.scheduler import WakeScheduler from traderai.uex_client import UEXClient ToolHandler = Callable[..., Awaitable[dict[str, Any]]] UEX_GET_RESOURCES: dict[str, dict[str, Any]] = { "categories": {"params": ["type", "section"], "auth": False, "group": "reference"}, "categories_attributes": {"params": ["id_category", "category_name", "category_type"], "auth": False, "group": "reference"}, "cities": {"params": ["id", "id_planet", "id_star_system", "name", "slug"], "auth": False, "group": "locations"}, "commodities": {"params": ["id", "name", "code", "slug"], "auth": False, "group": "trade"}, "commodities_alerts": {"params": ["id_commodity", "commodity_name", "commodity_code", "commodity_slug"], "auth": False, "group": "trade"}, "commodities_averages": {"params": ["id_commodity", "commodity_name", "commodity_code", "commodity_slug"], "auth": False, "group": "trade"}, "commodities_prices": { "params": ["id_terminal", "id_commodity", "terminal_name", "terminal_code", "terminal_slug", "commodity_name", "commodity_code", "commodity_slug"], "auth": False, "group": "trade", }, "commodities_prices_all": {"params": [], "auth": False, "group": "trade", "heavy": True}, "commodities_prices_history": {"params": ["id_terminal", "id_commodity", "game_version"], "auth": False, "group": "trade", "history": True}, "commodities_ranking": {"params": ["id_commodity", "commodity_name", "commodity_code", "commodity_slug"], "auth": False, "group": "trade"}, "commodities_raw_averages": {"params": ["id_commodity", "commodity_name", "commodity_code", "commodity_slug"], "auth": False, "group": "mining"}, "commodities_raw_prices": {"params": ["id_terminal", "id_commodity", "terminal_name", "commodity_name"], "auth": False, "group": "mining"}, "commodities_raw_prices_all": {"params": [], "auth": False, "group": "mining", "heavy": True}, "commodities_routes": {"params": ["id_terminal_origin", "id_terminal_destination", "id_commodity", "terminal_origin_name", "terminal_destination_name", "commodity_name"], "auth": False, "group": "trade"}, "commodities_status": {"params": [], "auth": False, "group": "trade"}, "companies": {"params": ["id", "name", "code"], "auth": False, "group": "reference"}, "contacts": {"params": ["id", "name"], "auth": False, "group": "reference"}, "contracts": {"params": ["id", "name", "slug"], "auth": False, "group": "reference"}, "crew": {"params": ["id", "name", "slug"], "auth": False, "group": "reference"}, "currencies_index": {"params": ["code"], "auth": False, "group": "reference"}, "currencies_index_history": {"params": ["currency", "date_from", "date_to"], "auth": False, "group": "reference", "history": True}, "data_extract": {"params": ["table"], "auth": False, "group": "data"}, "data_parameters": {"params": ["endpoint"], "auth": False, "group": "data"}, "factions": {"params": ["id", "name", "slug"], "auth": False, "group": "reference"}, "fleet": {"params": ["username"], "auth": False, "group": "vehicles"}, "fuel_prices": {"params": ["id_terminal", "terminal_name", "terminal_code", "terminal_slug"], "auth": False, "group": "trade"}, "fuel_prices_all": {"params": [], "auth": False, "group": "trade", "heavy": True}, "game_versions": {"params": [], "auth": False, "group": "reference"}, "items": {"params": ["id", "id_category", "name", "uuid", "slug"], "auth": False, "group": "items"}, "items_attributes": {"params": ["id_item", "item_name", "item_slug"], "auth": False, "group": "items"}, "items_prices": {"params": ["id_item", "id_terminal", "item_name", "terminal_name"], "auth": False, "group": "items"}, "items_prices_all": {"params": [], "auth": False, "group": "items", "heavy": True}, "jump_points": {"params": ["id", "name", "slug"], "auth": False, "group": "locations"}, "jurisdictions": {"params": ["id", "name"], "auth": False, "group": "locations"}, "marketplace_averages": {"params": ["id_item", "item_name", "item_slug"], "auth": False, "group": "marketplace"}, "marketplace_averages_all": {"params": [], "auth": False, "group": "marketplace", "heavy": True}, "marketplace_favorites": {"params": ["id_listing"], "auth": True, "group": "marketplace"}, "marketplace_listings": {"params": ["id", "slug", "username"], "auth": False, "group": "marketplace"}, "marketplace_negotiations": {"params": ["id", "id_listing", "hash"], "auth": True, "group": "marketplace"}, "marketplace_negotiations_messages": {"params": ["hash", "id_negotiation"], "auth": True, "group": "marketplace"}, "marketplace_prices_averages": {"params": ["id_item", "item_name", "item_slug"], "auth": False, "group": "marketplace"}, "marketplace_prices_averages_all": {"params": [], "auth": False, "group": "marketplace", "heavy": True}, "marketplace_prices_history": { "params": [ "id_item", "id_listing", "id_terminal", "id_star_system", "id_category", "item_uuid", "item_name", "operation", "quality_tier", "currency", "game_version", "date_start", "date_end", ], "auth": False, "group": "marketplace", "history": True, }, "marketplace_trends": {"params": ["id_item", "item_name", "item_slug"], "auth": False, "group": "marketplace"}, "moons": {"params": ["id", "id_planet", "id_star_system", "name", "slug"], "auth": False, "group": "locations"}, "orbits": {"params": ["id", "id_star_system", "name", "slug"], "auth": False, "group": "locations"}, "orbits_distances": {"params": ["id_origin", "id_destination"], "auth": False, "group": "locations"}, "organizations": {"params": ["sid", "name"], "auth": False, "group": "reference"}, "outposts": {"params": ["id", "id_moon", "id_planet", "name", "slug"], "auth": False, "group": "locations"}, "planets": {"params": ["id", "id_star_system", "name", "slug"], "auth": False, "group": "locations"}, "poi": {"params": ["id", "id_star_system", "name", "slug"], "auth": False, "group": "locations"}, "refineries_audits": {"params": ["id_terminal", "terminal_name"], "auth": False, "group": "mining"}, "refineries_capacities": {"params": ["id_terminal", "terminal_name"], "auth": False, "group": "mining"}, "refineries_methods": {"params": ["id", "name"], "auth": False, "group": "mining"}, "refineries_yields": {"params": ["id_terminal", "id_commodity", "terminal_name", "commodity_name"], "auth": False, "group": "mining"}, "release_notes": {"params": [], "auth": False, "group": "reference"}, "space_stations": {"params": ["id", "id_star_system", "id_planet", "id_moon", "name", "slug"], "auth": False, "group": "locations"}, "star_systems": {"params": ["id", "name", "code", "slug"], "auth": False, "group": "locations"}, "terminals": {"params": ["id", "id_star_system", "name", "code", "slug"], "auth": False, "group": "locations"}, "terminals_distances": {"params": ["id_terminal_origin", "id_terminal_destination"], "auth": False, "group": "locations"}, "user": {"params": ["username"], "auth": False, "group": "user"}, "user_notifications": {"params": [], "auth": True, "group": "user"}, "user_refineries_jobs": {"params": ["id"], "auth": True, "group": "user"}, "user_trades": {"params": ["id"], "auth": True, "group": "user"}, "vehicles": {"params": ["id", "name", "slug", "uuid"], "auth": False, "group": "vehicles"}, "vehicles_loaners": {"params": ["id_vehicle", "vehicle_name", "vehicle_slug"], "auth": False, "group": "vehicles"}, "vehicles_prices": {"params": ["id_vehicle", "vehicle_name", "vehicle_slug"], "auth": False, "group": "vehicles"}, "vehicles_purchases_prices": {"params": ["id_vehicle", "id_terminal", "vehicle_name", "terminal_name"], "auth": False, "group": "vehicles"}, "vehicles_purchases_prices_all": {"params": [], "auth": False, "group": "vehicles", "heavy": True}, "vehicles_rentals_prices": {"params": ["id_vehicle", "id_terminal", "vehicle_name", "terminal_name"], "auth": False, "group": "vehicles"}, "vehicles_rentals_prices_all": {"params": [], "auth": False, "group": "vehicles", "heavy": True}, "wallet_balance": {"params": [], "auth": True, "group": "user"}, } UEX_POST_RESOURCES = { "data_submit", "marketplace_advertise", "marketplace_negotiations_messages", "user_refineries_jobs_add", "user_trades_add", "user_trades_edit", "wallet_add", } UEX_DELETE_RESOURCES = { "marketplace_listings", "user_refineries_jobs_remove", "user_trades_remove", } UEX_RESOURCE_DESCRIPTIONS = { "commodities_prices_history": "Historical commodity prices at a terminal. Requires id_terminal and id_commodity; accepts game_version. UEX limits this to 500 rows.", "marketplace_prices_history": "Historical marketplace price snapshots, one row per listing per price change. Requires at least one filter; supports date_start/date_end and up to 1000 records.", "currencies_index_history": "Historical UEX currency index snapshots with basket component detail. Supports currency, date_from, and date_to timestamps.", } @dataclass class PendingAction: id: str label: str endpoint: str payload: dict[str, Any] method: str = "POST" class ToolRegistry: def __init__( self, uex: UEXClient, require_write_approval: bool = True, memory: MemoryStore | None = None, scheduler: WakeScheduler | None = None, ) -> None: self.uex = uex self.require_write_approval = require_write_approval self.memory = memory self.scheduler = scheduler self.pending_actions: dict[str, PendingAction] = {} self.handlers: dict[str, ToolHandler] = { "search_marketplace_listings": self.search_marketplace_listings, "get_marketplace_listing": self.get_marketplace_listing, "list_marketplace_negotiations": self.list_marketplace_negotiations, "get_negotiation_messages": self.get_negotiation_messages, "draft_negotiation_message": self.draft_negotiation_message, "draft_marketplace_listing": self.draft_marketplace_listing, "remember_user_fact": self.remember_user_fact, "recall_memory": self.recall_memory, "schedule_wake_job": self.schedule_wake_job, "list_wake_jobs": self.list_wake_jobs, "check_uex_notifications": self.check_uex_notifications, } self.handlers["uex_api_catalog"] = self.uex_api_catalog self.handlers["uex_get"] = self.uex_get self.handlers["uex_draft_post"] = self.uex_draft_post self.handlers["uex_draft_delete"] = self.uex_draft_delete self.handlers["search_uex_api_index"] = self.search_uex_api_index self.handlers["summarize_uex_commodity_price_history"] = self.summarize_uex_commodity_price_history self.handlers["summarize_uex_marketplace_price_history"] = self.summarize_uex_marketplace_price_history self.handlers["summarize_uex_currency_index_history"] = self.summarize_uex_currency_index_history for resource in UEX_GET_RESOURCES: self.handlers[self._get_tool_name(resource)] = self._make_get_handler(resource) for resource in UEX_POST_RESOURCES: self.handlers[self._post_tool_name(resource)] = self._make_post_handler(resource) for resource in UEX_DELETE_RESOURCES: self.handlers[self._delete_tool_name(resource)] = self._make_delete_handler(resource) @property def schemas(self) -> list[dict[str, Any]]: return [ self._api_index_schema(), *self._uex_get_schemas(), *self._history_summary_schemas(), *self._uex_post_schemas(), *self._uex_delete_schemas(), { "type": "function", "function": { "name": "search_marketplace_listings", "description": "Search active/current UEX marketplace listings only. Prices are in-game aUEC/UEC credits, not real-world dollars. Do not use this as historical sale or completed-sale information. UEX returns up to 100 active listings; filters are applied locally.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Text to search in title, description, location, advertiser, or slug."}, "operation": {"type": "string", "enum": ["buy", "sell"]}, "type": {"type": "string", "enum": ["item", "service", "contract"]}, "username": {"type": "string", "description": "Advertiser IGN."}, "location": {"type": "string"}, "min_price": {"type": "number"}, "max_price": {"type": "number"}, "limit": {"type": "integer", "minimum": 1, "maximum": 25}, }, }, }, }, { "type": "function", "function": { "name": "get_marketplace_listing", "description": "Fetch a specific UEX marketplace listing by id or slug.", "parameters": { "type": "object", "properties": { "id": {"type": "integer"}, "slug": {"type": "string"}, }, }, }, }, { "type": "function", "function": { "name": "list_marketplace_negotiations", "description": "List authenticated marketplace negotiations for the configured UEX user.", "parameters": { "type": "object", "properties": { "id": {"type": "integer"}, "id_listing": {"type": "integer"}, "hash": {"type": "string"}, }, }, }, }, { "type": "function", "function": { "name": "get_negotiation_messages", "description": "Fetch authenticated messages from a marketplace negotiation by hash or id_negotiation.", "parameters": { "type": "object", "properties": { "hash": {"type": "string"}, "id_negotiation": {"type": "integer"}, }, }, }, }, { "type": "function", "function": { "name": "draft_negotiation_message", "description": "Draft a message or offer to a UEX negotiation. This creates a pending action that must be approved before sending.", "parameters": { "type": "object", "required": ["message"], "properties": { "message": {"type": "string"}, "hash": {"type": "string"}, "id_negotiation": {"type": "integer"}, "is_production": {"type": "integer", "enum": [0, 1], "default": 0}, }, }, }, }, { "type": "function", "function": { "name": "draft_marketplace_listing", "description": "Draft a new UEX marketplace listing. Listing prices are in-game aUEC/UEC credits, not real-world dollars. This creates a pending action that must be approved before posting.", "parameters": { "type": "object", "required": ["id_category", "operation", "type", "unit", "title", "description", "price", "currency", "language"], "properties": { "id_item": {"type": "integer"}, "id_star_system": {"type": "integer"}, "id_organization": {"type": "integer"}, "id_category": {"type": "integer"}, "operation": {"type": "string", "enum": ["buy", "sell"]}, "type": {"type": "string", "enum": ["item", "service", "contract"]}, "unit": {"type": "string"}, "title": {"type": "string"}, "description": {"type": "string"}, "price": {"type": "number"}, "currency": {"type": "string", "enum": ["UEC", "WIF"]}, "language": {"type": "string", "default": "en_US"}, "location": {"type": "string"}, "source": {"type": "string"}, "availability": {"type": "string"}, "in_stock": {"type": "integer"}, "hours_expiration": {"type": "integer"}, "is_hidden": {"type": "integer", "enum": [0, 1]}, "is_production": {"type": "integer", "enum": [0, 1], "default": 0}, }, }, }, }, { "type": "function", "function": { "name": "remember_user_fact", "description": "Persist a durable user preference, identity detail, trading rule, or long-term note for future chats.", "parameters": { "type": "object", "required": ["content"], "properties": { "content": {"type": "string"}, "kind": {"type": "string", "enum": ["user", "preference", "trading", "project", "note"], "default": "note"}, "importance": {"type": "integer", "minimum": 1, "maximum": 5, "default": 3}, }, }, }, }, { "type": "function", "function": { "name": "recall_memory", "description": "Search long-term memory for relevant prior facts, preferences, and chat context.", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer", "minimum": 1, "maximum": 10, "default": 6}, }, }, }, }, { "type": "function", "function": { "name": "schedule_wake_job", "description": "Create a scheduled wake-up job for the assistant. Use either run_at for one-time jobs or cron for recurring jobs.", "parameters": { "type": "object", "required": ["prompt"], "properties": { "prompt": {"type": "string", "description": "What the AI should consider or do when it wakes."}, "run_at": {"type": "string", "description": "ISO datetime for a one-time wake job, such as 2026-05-05T20:30:00-04:00."}, "cron": {"type": "string", "description": "Five-field cron expression for recurring jobs, such as 0 9 * * *."}, }, }, }, }, { "type": "function", "function": { "name": "list_wake_jobs", "description": "List currently enabled scheduled assistant wake jobs.", "parameters": {"type": "object", "properties": {}}, }, }, { "type": "function", "function": { "name": "check_uex_notifications", "description": "Check authenticated UEX user notifications and return unread pending notifications.", "parameters": {"type": "object", "properties": {}}, }, }, ] async def execute(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]: handler = self.handlers.get(name) if not handler: return {"error": f"Unknown tool: {name}"} try: return await handler(**arguments) except Exception as exc: return {"error": str(exc)} async def approve(self, action_id: str) -> dict[str, Any]: action = self.pending_actions.pop(action_id, None) if not action: return {"error": f"Pending action not found: {action_id}"} if action.method == "DELETE": return await self.uex.delete(action.endpoint, action.payload, authenticated=True) return await self.uex.post(action.endpoint, action.payload, authenticated=True) async def decline(self, action_id: str) -> dict[str, Any]: action = self.pending_actions.pop(action_id, None) if not action: return {"error": f"Pending action not found: {action_id}"} return { "declined": True, "pending_action": { "id": action.id, "label": action.label, "method": action.method, "endpoint": action.endpoint, "payload": action.payload, }, } async def uex_api_catalog(self, group: str | None = None, resource: str | None = None) -> dict[str, Any]: if resource: key = self._validate_resource(resource, UEX_GET_RESOURCES) info = UEX_GET_RESOURCES[key] return { "resource": key, "method": "GET", "group": info["group"], "authenticated": info["auth"], "heavy": bool(info.get("heavy")), "params": info["params"], "write_resources": { "post": sorted(UEX_POST_RESOURCES), "delete": sorted(UEX_DELETE_RESOURCES), }, } grouped: dict[str, list[dict[str, Any]]] = {} for name, info in sorted(UEX_GET_RESOURCES.items()): if group and info["group"] != group: continue grouped.setdefault(info["group"], []).append( { "resource": name, "params": info["params"], "auth": info["auth"], "heavy": bool(info.get("heavy")), } ) return { "get": grouped, "post": sorted(UEX_POST_RESOURCES), "delete": sorted(UEX_DELETE_RESOURCES), "usage": "Call uex_get(resource, params, fields, limit, mode). Use fields and limit to keep responses small.", } async def uex_get( self, resource: str, params: dict[str, Any] | None = None, fields: list[str] | None = None, search: str | None = None, limit: int = 10, offset: int = 0, mode: str = "summary", ) -> dict[str, Any]: resource = self._validate_resource(resource, UEX_GET_RESOURCES) info = UEX_GET_RESOURCES[resource] cleaned_params = self._filter_params(params or {}, info["params"]) response = await self.uex.get(resource, cleaned_params, authenticated=bool(info["auth"])) data = response.get("data") items = self._as_list(data) total = len(items) if search: needle = search.casefold() items = [item for item in items if needle in self._search_text(item)] filtered_total = len(items) offset = max(0, offset) limit = max(1, min(limit, 100)) window = items[offset : offset + limit] compacted = [ self._project_item(item, fields=fields, mode=mode) for item in window ] return { "status": response.get("status"), "resource": resource, "params": cleaned_params, "total": total, "matched": filtered_total, "returned": len(compacted), "offset": offset, "truncated": offset + len(compacted) < filtered_total, "items": compacted, } async def uex_draft_post(self, resource: str, payload: dict[str, Any], label: str | None = None) -> dict[str, Any]: resource = self._validate_resource(resource, UEX_POST_RESOURCES) return self._pending(label or f"POST {resource}", resource, payload, method="POST") async def uex_draft_delete( self, resource: str, params: dict[str, Any] | None = None, label: str | None = None, ) -> dict[str, Any]: resource = self._validate_resource(resource, UEX_DELETE_RESOURCES) return self._pending(label or f"DELETE {resource}", resource, params or {}, method="DELETE") async def search_uex_api_index( self, query: str = "", group: str | None = None, history_only: bool = False, limit: int = 20, ) -> dict[str, Any]: needle = query.casefold().strip() matches = [] for resource, info in sorted(UEX_GET_RESOURCES.items()): if group and info["group"] != group: continue if history_only and not info.get("history"): continue haystack = " ".join( [ resource, info["group"], " ".join(info["params"]), UEX_RESOURCE_DESCRIPTIONS.get(resource, ""), ] ).casefold() if needle and needle not in haystack: continue matches.append(self._resource_index_entry("GET", resource, info)) if len(matches) >= max(1, min(limit, 50)): break post_matches = [] if not history_only: for resource in sorted(UEX_POST_RESOURCES): if group and group != "write": continue if needle and needle not in resource.casefold(): continue post_matches.append( { "method": "POST", "resource": resource, "tool": self._post_tool_name(resource), "approval_required": True, "docs_url": self._docs_url("post", resource), } ) delete_matches = [] if not history_only: for resource in sorted(UEX_DELETE_RESOURCES): if group and group != "write": continue if needle and needle not in resource.casefold(): continue delete_matches.append( { "method": "DELETE", "resource": resource, "tool": self._delete_tool_name(resource), "approval_required": True, "docs_url": self._docs_url("delete", resource), } ) return { "count": len(matches) + len(post_matches) + len(delete_matches), "get": matches, "post": post_matches[: max(0, min(limit, 50) - len(matches))], "delete": delete_matches[: max(0, min(limit, 50) - len(matches) - len(post_matches))], } async def summarize_uex_commodity_price_history( self, id_terminal: int, id_commodity: int, game_version: str | None = None, limit: int = 100, ) -> dict[str, Any]: return await self._history_summary( "commodities_prices_history", {"id_terminal": id_terminal, "id_commodity": id_commodity, "game_version": game_version}, value_fields=["price_buy", "price_sell", "scu_buy", "scu_sell", "scu_sell_stock"], label_fields=["commodity_name", "terminal_name", "game_version"], limit=limit, ) async def summarize_uex_marketplace_price_history( self, id_item: str | int | None = None, id_listing: int | None = None, id_terminal: int | None = None, id_star_system: int | None = None, id_category: int | None = None, item_uuid: str | None = None, item_name: str | None = None, operation: str | None = None, quality_tier: int | None = None, currency: str | None = None, game_version: str | None = None, date_start: str | None = None, date_end: str | None = None, limit: int = 250, ) -> dict[str, Any]: params = { "id_item": id_item, "id_listing": id_listing, "id_terminal": id_terminal, "id_star_system": id_star_system, "id_category": id_category, "item_uuid": item_uuid, "item_name": item_name, "operation": operation, "quality_tier": quality_tier, "currency": currency, "game_version": game_version, "date_start": date_start, "date_end": date_end, } return await self._history_summary( "marketplace_prices_history", params, value_fields=["price", "quality"], label_fields=["item_name", "operation", "currency", "terminal_name", "game_version"], limit=limit, ) async def summarize_uex_currency_index_history( self, currency: str | None = None, date_from: int | None = None, date_to: int | None = None, limit: int = 365, ) -> dict[str, Any]: return await self._history_summary( "currencies_index_history", {"currency": currency, "date_from": date_from, "date_to": date_to}, value_fields=["index_value", "basket_value", "data_window_days"], label_fields=["currency", "methodology"], limit=limit, ) def _make_get_handler(self, resource: str) -> ToolHandler: async def handler(**arguments: Any) -> dict[str, Any]: fields = arguments.pop("fields", None) search = arguments.pop("search", None) limit = arguments.pop("limit", 10) offset = arguments.pop("offset", 0) mode = arguments.pop("mode", "summary") return await self.uex_get( resource, params=arguments, fields=fields, search=search, limit=limit, offset=offset, mode=mode, ) return handler def _make_post_handler(self, resource: str) -> ToolHandler: async def handler(payload: dict[str, Any], label: str | None = None) -> dict[str, Any]: return await self.uex_draft_post(resource, payload, label=label) return handler def _make_delete_handler(self, resource: str) -> ToolHandler: async def handler(label: str | None = None, **params: Any) -> dict[str, Any]: return await self.uex_draft_delete(resource, params, label=label) return handler @classmethod def _uex_get_schemas(cls) -> list[dict[str, Any]]: return [ { "type": "function", "function": { "name": cls._get_tool_name(resource), "description": cls._get_tool_description(resource, info), "parameters": cls._get_tool_parameters(info["params"]), }, } for resource, info in sorted(UEX_GET_RESOURCES.items()) ] @classmethod def _api_index_schema(cls) -> dict[str, Any]: return { "type": "function", "function": { "name": "search_uex_api_index", "description": "Search the indexed UEX API tool catalog by topic, resource, parameter, or group. Use to discover exact tool names, especially history tools.", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "group": { "type": "string", "enum": ["trade", "marketplace", "items", "vehicles", "locations", "mining", "user", "reference", "data", "write"], }, "history_only": {"type": "boolean", "default": False}, "limit": {"type": "integer", "minimum": 1, "maximum": 50, "default": 20}, }, }, }, } @classmethod def _history_summary_schemas(cls) -> list[dict[str, Any]]: controls = { "limit": {"type": "integer", "minimum": 1, "maximum": 1000, "default": 250}, } return [ { "type": "function", "function": { "name": "summarize_uex_commodity_price_history", "description": "Summarize historical commodity price and inventory changes for one commodity at one terminal.", "parameters": { "type": "object", "required": ["id_terminal", "id_commodity"], "properties": { "id_terminal": {"type": "integer"}, "id_commodity": {"type": "integer"}, "game_version": {"type": "string"}, **controls, }, }, }, }, { "type": "function", "function": { "name": "summarize_uex_marketplace_price_history", "description": "Summarize marketplace historical price snapshots for an item, listing, terminal, category, system, or date range.", "parameters": { "type": "object", "properties": { "id_item": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "id_listing": {"type": "integer"}, "id_terminal": {"type": "integer"}, "id_star_system": {"type": "integer"}, "id_category": {"type": "integer"}, "item_uuid": {"type": "string"}, "item_name": {"type": "string"}, "operation": {"type": "string", "enum": ["buy", "sell"]}, "quality_tier": {"type": "integer", "minimum": 0, "maximum": 4}, "currency": {"type": "string"}, "game_version": {"type": "string"}, "date_start": {"type": "string", "description": "YYYY-MM-DD"}, "date_end": {"type": "string", "description": "YYYY-MM-DD"}, **controls, }, }, }, }, { "type": "function", "function": { "name": "summarize_uex_currency_index_history", "description": "Summarize historical UEX currency index snapshots and basket value changes.", "parameters": { "type": "object", "properties": { "currency": {"type": "string"}, "date_from": {"type": "integer", "description": "Unix timestamp."}, "date_to": {"type": "integer", "description": "Unix timestamp."}, **controls, }, }, }, }, ] @classmethod def _uex_post_schemas(cls) -> list[dict[str, Any]]: return [ { "type": "function", "function": { "name": cls._post_tool_name(resource), "description": f"Draft UEX POST /{resource}/ for user approval. Nothing is sent until approval.", "parameters": { "type": "object", "required": ["payload"], "properties": { "payload": {"type": "object", "description": f"JSON body for UEX POST /{resource}/."}, "label": {"type": "string", "description": "Short approval label."}, }, }, }, } for resource in sorted(UEX_POST_RESOURCES) ] @classmethod def _uex_delete_schemas(cls) -> list[dict[str, Any]]: return [ { "type": "function", "function": { "name": cls._delete_tool_name(resource), "description": f"Draft UEX DELETE /{resource}/ for user approval. Nothing is deleted until approval.", "parameters": { "type": "object", "properties": { "id": {"type": "integer"}, "label": {"type": "string", "description": "Short approval label."}, }, }, }, } for resource in sorted(UEX_DELETE_RESOURCES) ] @classmethod def _get_tool_parameters(cls, endpoint_params: list[str]) -> dict[str, Any]: properties = { param: cls._query_param_schema(param) for param in endpoint_params } properties.update( { "fields": { "type": "array", "items": {"type": "string"}, "description": "Fields to keep in each result row.", }, "search": {"type": "string", "description": "Local text filter after UEX returns data."}, "limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 10}, "offset": {"type": "integer", "minimum": 0, "default": 0}, "mode": {"type": "string", "enum": ["summary", "full"], "default": "summary"}, } ) return {"type": "object", "properties": properties} @staticmethod def _query_param_schema(param: str) -> dict[str, Any]: if param == "id" or param.startswith("id_") or param in {"date_from", "date_to", "quality_tier"}: return {"type": "integer"} return {"type": "string"} @staticmethod def _get_tool_description(resource: str, info: dict[str, Any]) -> str: auth = " Authenticated." if info["auth"] else "" heavy = " Heavy endpoint; use fields and limit." if info.get("heavy") else "" history = " History endpoint." if info.get("history") else "" description = UEX_RESOURCE_DESCRIPTIONS.get(resource) if description: return f"GET UEX /{resource}/ with compact, token-limited results. {description}{auth}{heavy}" return f"GET UEX /{resource}/ with compact, token-limited results.{history}{auth}{heavy}" @staticmethod def _get_tool_name(resource: str) -> str: return f"get_uex_{resource}" @staticmethod def _post_tool_name(resource: str) -> str: return f"draft_uex_{resource}" @staticmethod def _delete_tool_name(resource: str) -> str: return f"delete_uex_{resource}" @classmethod def _resource_index_entry(cls, method: str, resource: str, info: dict[str, Any]) -> dict[str, Any]: return { "method": method, "resource": resource, "tool": cls._get_tool_name(resource), "group": info["group"], "params": info["params"], "authenticated": info["auth"], "history": bool(info.get("history")), "heavy": bool(info.get("heavy")), "description": UEX_RESOURCE_DESCRIPTIONS.get(resource, ""), "docs_url": cls._docs_url("get", resource), } @staticmethod def _docs_url(method: str, resource: str) -> str: return f"https://uexcorp.space/api/documentation/id/{method}_{resource}/" async def search_marketplace_listings( self, query: str | None = None, operation: str | None = None, type: str | None = None, username: str | None = None, location: str | None = None, min_price: float | None = None, max_price: float | None = None, limit: int = 10, ) -> dict[str, Any]: response = await self.uex.get("marketplace_listings", {"username": username}) listings = response.get("data") or [] filtered = [] q = (query or "").casefold() loc = (location or "").casefold() for listing in listings: if operation and listing.get("operation") != operation: continue if type and listing.get("type") != type: continue if min_price is not None and float(listing.get("price") or 0) < min_price: continue if max_price is not None and float(listing.get("price") or 0) > max_price: continue if loc and loc not in str(listing.get("location") or "").casefold(): continue haystack = " ".join(str(listing.get(k) or "") for k in ["title", "description", "location", "user_username", "slug"]).casefold() if q and q not in haystack: continue filtered.append(self._summarize_listing(listing)) if len(filtered) >= max(1, min(limit, 25)): break return {"count": len(filtered), "listings": filtered} async def get_marketplace_listing(self, id: int | None = None, slug: str | None = None) -> dict[str, Any]: response = await self.uex.get("marketplace_listings", {"id": id, "slug": slug}) return {"listing": response.get("data")} async def list_marketplace_negotiations( self, id: int | None = None, id_listing: int | None = None, hash: str | None = None, ) -> dict[str, Any]: return await self.uex.get("marketplace_negotiations", {"id": id, "id_listing": id_listing, "hash": hash}, authenticated=True) async def get_negotiation_messages(self, hash: str | None = None, id_negotiation: int | None = None) -> dict[str, Any]: return await self.uex.get("marketplace_negotiations_messages", {"hash": hash, "id_negotiation": id_negotiation}, authenticated=True) async def draft_negotiation_message( self, message: str, hash: str | None = None, id_negotiation: int | None = None, is_production: int = 0, ) -> dict[str, Any]: payload = {"message": message, "hash": hash, "id_negotiation": id_negotiation, "is_production": is_production} return self._pending("Send negotiation message", "marketplace_negotiations_messages", payload) async def draft_marketplace_listing(self, **payload: Any) -> dict[str, Any]: return self._pending("Post marketplace listing", "marketplace_advertise", payload) async def remember_user_fact(self, content: str, kind: str = "note", importance: int = 3) -> dict[str, Any]: if self.memory is None: return {"error": "Memory store is not configured."} return {"memory": self.memory.remember(kind, content, importance)} async def recall_memory(self, query: str = "", limit: int = 6) -> dict[str, Any]: if self.memory is None: return {"error": "Memory store is not configured."} return {"memories": self.memory.recall(query, max(1, min(limit, 10)))} async def schedule_wake_job( self, prompt: str, run_at: str | None = None, cron: str | None = None, ) -> dict[str, Any]: if self.scheduler is None: return {"error": "Scheduler is not configured."} if bool(run_at) == bool(cron): return {"error": "Provide exactly one of run_at or cron."} if run_at: return {"scheduled_job": self.scheduler.schedule_date(run_at, prompt)} return {"scheduled_job": self.scheduler.schedule_cron(cron or "", prompt)} async def list_wake_jobs(self) -> dict[str, Any]: if self.scheduler is None: return {"error": "Scheduler is not configured."} return {"scheduled_jobs": self.scheduler.list_jobs()} async def check_uex_notifications(self) -> dict[str, Any]: response = await self.uex.get_user_notifications() notifications = response.get("notifications") or [] pending = [item for item in notifications if not item.get("date_read")] return {"count": len(pending), "notifications": pending} def _pending(self, label: str, endpoint: str, payload: dict[str, Any], method: str = "POST") -> dict[str, Any]: action_id = str(uuid.uuid4()) payload = {key: value for key, value in payload.items() if value is not None} self.pending_actions[action_id] = PendingAction(action_id, label, endpoint, payload, method) return { "pending_action": { "id": action_id, "label": label, "method": method, "endpoint": endpoint, "payload": payload, "approval_required": self.require_write_approval, } } @staticmethod def _validate_resource(resource: str, allowed: dict[str, Any] | set[str]) -> str: normalized = resource.strip().strip("/").casefold() if normalized not in allowed: choices = sorted(allowed.keys() if isinstance(allowed, dict) else allowed) near = [name for name in choices if normalized in name or name in normalized][:8] hint = f" Did you mean: {', '.join(near)}?" if near else "" raise ValueError(f"Unsupported UEX resource: {resource}.{hint}") return normalized @staticmethod def _filter_params(params: dict[str, Any], allowed_params: list[str]) -> dict[str, Any]: if not allowed_params: return {key: value for key, value in params.items() if value is not None} allowed = set(allowed_params) return {key: value for key, value in params.items() if key in allowed and value is not None} @staticmethod def _as_list(data: Any) -> list[Any]: if data is None: return [] if isinstance(data, list): return data return [data] @classmethod def _project_item(cls, item: Any, fields: list[str] | None = None, mode: str = "summary") -> Any: if not isinstance(item, dict): return item if fields: return {field: cls._compact_scalar(item.get(field)) for field in fields if field in item} if mode == "full": return {key: cls._compact_scalar(value) for key, value in item.items()} priority = [ "id", "uuid", "code", "slug", "name", "title", "type", "section", "operation", "price", "currency", "unit", "location", "terminal_name", "commodity_name", "item_name", "vehicle_name", "price_buy", "price_sell", "scu_buy", "scu_sell", "scu_sell_stock", "status_buy", "status_sell", "date_modified", "date_added", ] selected: dict[str, Any] = {} for key in priority: if key in item and item[key] not in (None, ""): selected[key] = cls._compact_scalar(item[key]) for key, value in item.items(): if len(selected) >= 16: break if key in selected or value in (None, ""): continue if isinstance(value, (str, int, float, bool)): selected[key] = cls._compact_scalar(value) return selected @staticmethod def _compact_scalar(value: Any) -> Any: if isinstance(value, str) and len(value) > 240: return value[:237] + "..." if isinstance(value, list): return value[:5] if isinstance(value, dict): return {key: nested_value for key, nested_value in list(value.items())[:12]} return value @classmethod def _search_text(cls, item: Any) -> str: if isinstance(item, dict): return " ".join(str(value) for value in item.values() if isinstance(value, (str, int, float))).casefold() return str(item).casefold() async def _history_summary( self, resource: str, params: dict[str, Any], value_fields: list[str], label_fields: list[str], limit: int, ) -> dict[str, Any]: info = UEX_GET_RESOURCES[resource] cleaned_params = self._filter_params(params, info["params"]) response = await self.uex.get(resource, cleaned_params, authenticated=bool(info["auth"])) rows = [ row for row in self._as_list(response.get("data")) if isinstance(row, dict) ][: max(1, min(limit, 1000))] rows_sorted = sorted(rows, key=lambda row: int(row.get("date_added") or 0)) latest = rows_sorted[-1] if rows_sorted else {} earliest = rows_sorted[0] if rows_sorted else {} summaries = { field: self._numeric_history_summary(rows_sorted, field) for field in value_fields if any(self._is_number(row.get(field)) for row in rows_sorted) } labels = { field: latest.get(field) for field in label_fields if latest.get(field) not in (None, "") } sample_fields = ["id", "date_added", *label_fields, *value_fields] recent = [ self._project_item(row, fields=sample_fields, mode="summary") for row in list(reversed(rows_sorted[-5:])) ] return { "status": response.get("status"), "resource": resource, "params": cleaned_params, "count": len(rows), "date_start": earliest.get("date_added"), "date_end": latest.get("date_added"), "labels": labels, "metrics": summaries, "recent": recent, "docs_url": self._docs_url("get", resource), } @classmethod def _numeric_history_summary(cls, rows: list[dict[str, Any]], field: str) -> dict[str, Any]: points = [ (int(row.get("date_added") or 0), float(row[field])) for row in rows if cls._is_number(row.get(field)) ] values = [value for _, value in points] first_date, first_value = points[0] last_date, last_value = points[-1] change = last_value - first_value pct_change = (change / first_value * 100) if first_value else None return { "first": first_value, "first_date": first_date, "latest": last_value, "latest_date": last_date, "min": min(values), "max": max(values), "avg": round(sum(values) / len(values), 4), "change": round(change, 4), "pct_change": round(pct_change, 4) if pct_change is not None else None, "points": len(points), } @staticmethod def _is_number(value: Any) -> bool: return isinstance(value, (int, float)) and not isinstance(value, bool) @staticmethod def _summarize_listing(listing: dict[str, Any]) -> dict[str, Any]: return { "id": listing.get("id"), "slug": listing.get("slug"), "title": listing.get("title"), "operation": listing.get("operation"), "type": listing.get("type"), "price": listing.get("price"), "currency": listing.get("currency"), "unit": listing.get("unit"), "location": listing.get("location"), "availability": listing.get("availability"), "in_stock": listing.get("in_stock"), "advertiser": listing.get("user_username"), "expires_at": listing.get("date_expiration"), }