from __future__ import annotations import uuid from contextlib import contextmanager from contextvars import ContextVar from dataclasses import dataclass from typing import Any, Awaitable, Callable from traderai.cornerstone_client import CornerstoneClient, parse_cornerstone_item_page from traderai.memory import MemoryStore from traderai.negotiations import UEX_NEGOTIATION_CLOSE_ENDPOINT from traderai.scheduler import WakeScheduler from traderai.scmdb_client import SCMDBClient from traderai.starcitizen_wiki_client import StarCitizenWikiClient from traderai.uex_client import UEXClient from traderai.wikelo_projects_client import WikeloProjectsClient 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", "id_item", "operation"], "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", "id_category", "currency", "quality_tier"], "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", "id_category", "currency", "quality_tier"], "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.", "marketplace_trends": "Current UEX marketplace trend metrics for an item. Use this when the user asks for trends, price movement, demand, or what the market is doing now.", "currencies_index_history": "Historical UEX currency index snapshots with basket component detail. Supports currency, date_from, and date_to timestamps.", } UEX_PRODUCTION_WRITE_RESOURCES = { "marketplace_advertise", "marketplace_negotiations_messages", UEX_NEGOTIATION_CLOSE_ENDPOINT, } @dataclass class PendingAction: id: str label: str endpoint: str payload: dict[str, Any] method: str = "POST" metadata: dict[str, Any] | None = None class ToolRegistry: def __init__( self, uex: UEXClient, require_write_approval: bool = True, memory: MemoryStore | None = None, scheduler: WakeScheduler | None = None, scmdb: SCMDBClient | None = None, cornerstone: CornerstoneClient | None = None, scwiki: StarCitizenWikiClient | None = None, wikelo: WikeloProjectsClient | None = None, plan_store: Any | None = None, plan_runner: Any | None = None, negotiation_sync: Any | None = None, ) -> None: self.uex = uex self.scmdb = scmdb or SCMDBClient() self.cornerstone = cornerstone or CornerstoneClient() self.scwiki = scwiki or StarCitizenWikiClient() self.wikelo = wikelo or WikeloProjectsClient() self.require_write_approval = require_write_approval self.memory = memory self.scheduler = scheduler self.plan_store = plan_store self.plan_runner = plan_runner self.negotiation_sync = negotiation_sync self.pending_actions: dict[str, PendingAction] = {} self._chat_images_var: ContextVar[list[dict[str, Any]]] = ContextVar("chat_images", default=[]) self.handlers: dict[str, ToolHandler] = { "search_marketplace_listings": self.search_marketplace_listings, "get_marketplace_listing": self.get_marketplace_listing, "get_marketplace_trends": self.get_marketplace_trends, "list_marketplace_negotiations": self.list_marketplace_negotiations, "get_negotiation_messages": self.get_negotiation_messages, "draft_negotiation_message": self.draft_negotiation_message, "list_local_negotiations": self.list_local_negotiations, "get_local_negotiation": self.get_local_negotiation, "search_local_negotiation_messages": self.search_local_negotiation_messages, "draft_negotiation_close": self.draft_negotiation_close, "draft_negotiation_rating": self.draft_negotiation_rating, "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, "create_continual_plan": self.create_continual_plan, "list_continual_plans": self.list_continual_plans, "get_continual_plan": self.get_continual_plan, "pause_continual_plan": self.pause_continual_plan, "resume_continual_plan": self.resume_continual_plan, "cancel_continual_plan": self.cancel_continual_plan, "delete_continual_plan": self.delete_continual_plan, "run_continual_plan_now": self.run_continual_plan_now, "check_uex_notifications": self.check_uex_notifications, "list_scmdb_versions": self.list_scmdb_versions, "search_scmdb_missions": self.search_scmdb_missions, "get_scmdb_mission_rewards": self.get_scmdb_mission_rewards, "search_scwiki_pages": self.search_scwiki_pages, "get_scwiki_page": self.get_scwiki_page, "search_scwiki_vehicles": self.search_scwiki_vehicles, "get_scwiki_vehicle": self.get_scwiki_vehicle, "search_wikelo_ship_projects": self.search_wikelo_ship_projects, "get_wikelo_ship_project": self.get_wikelo_ship_project, "search_cornerstone_items": self.search_cornerstone_items, "get_cornerstone_item_locations": self.get_cornerstone_item_locations, "get_cornerstone_item_media": self.get_cornerstone_item_media, "draft_marketplace_listing_with_cornerstone_image": self.draft_marketplace_listing_with_cornerstone_image, } 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(), *self._scmdb_schemas(), *self._scwiki_schemas(), *self._wikelo_schemas(), *self._cornerstone_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": "get_marketplace_trends", "description": "Fetch current UEX marketplace trend metrics for an item, including WTS and WTB averages plus negotiation counts.", "parameters": { "type": "object", "properties": { "id_item": {"type": "integer"}, "item_name": {"type": "string"}, "item_slug": {"type": "string"}, "id_category": {"type": "integer"}, "currency": {"type": "string", "description": "Optional currency filter such as UEC, WIF, or MGS."}, "quality_tier": {"type": "integer", "minimum": 0, "maximum": 7}, }, }, }, }, { "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"}, "id_listing": {"type": "integer"}, "plan_id": {"type": "string"}, "plan_item_id": {"type": "integer"}, "candidate_id": {"type": "integer"}, "listing_slug": {"type": "string"}, "is_production": {"type": "integer", "enum": [0, 1], "default": 1}, }, }, }, }, { "type": "function", "function": { "name": "list_local_negotiations", "description": "List locally synced UEX negotiations with unread and status details.", "parameters": { "type": "object", "properties": { "status": {"type": "string", "enum": ["all", "open", "closed"]}, "unread_only": {"type": "boolean"}, "search": {"type": "string"}, "limit": {"type": "integer", "minimum": 1, "maximum": 50}, }, }, }, }, { "type": "function", "function": { "name": "get_local_negotiation", "description": "Get a locally synced UEX negotiation with compact metadata and recent messages.", "parameters": { "type": "object", "properties": { "hash": {"type": "string"}, }, "required": ["hash"], }, }, }, { "type": "function", "function": { "name": "search_local_negotiation_messages", "description": "Search locally cached negotiation message text so the assistant can reference prior UEX conversations without re-fetching them.", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer", "minimum": 1, "maximum": 20}, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "draft_negotiation_close", "description": "Draft closing or rating a UEX negotiation. This creates a pending action that must be approved before sending.", "parameters": { "type": "object", "properties": { "hash": {"type": "string"}, "id_negotiation": {"type": "integer"}, "deal_closed": {"type": "boolean"}, "deal_value": {"type": "number"}, "currency": {"type": "string"}, "clarity_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "speed_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "respect_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "fairness_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "comment": {"type": "string"}, }, "required": ["deal_closed"], }, }, }, { "type": "function", "function": { "name": "draft_negotiation_rating", "description": "Alias for drafting a UEX negotiation close/rating action.", "parameters": { "type": "object", "properties": { "hash": {"type": "string"}, "id_negotiation": {"type": "integer"}, "deal_closed": {"type": "boolean"}, "deal_value": {"type": "number"}, "currency": {"type": "string"}, "clarity_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "speed_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "respect_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "fairness_rating": {"type": "integer", "minimum": 1, "maximum": 5}, "comment": {"type": "string"}, }, "required": ["deal_closed"], }, }, }, { "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. Prefer draft_marketplace_listing_with_cornerstone_image for item posts when a Cornerstone image is useful.", "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"}, "durability": {"type": "integer", "minimum": 0, "maximum": 100}, "video_url": {"type": "string"}, "image_data": {"type": "string", "description": "Base64 JPG or PNG image data for UEX upload."}, "use_attached_image": { "type": "boolean", "description": "When true, reuse an image pasted into the current chat as the listing image_data.", }, "attached_image_index": { "type": "integer", "minimum": 0, "description": "Zero-based pasted image index to reuse when use_attached_image is true.", }, "hours_expiration": {"type": "integer"}, "is_hidden": {"type": "integer", "enum": [0, 1]}, "is_tv_allowed": {"type": "integer", "enum": [0, 1]}, "is_production": {"type": "integer", "enum": [0, 1], "default": 1}, }, }, }, }, { "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": "create_continual_plan", "description": "Create a durable multi-run plan. Use this for long-running marketplace work over days. kind=buying uses structured listing/candidate tracking; kind=custom continues through an agent wake prompt. All UEX writes are draft-only for approval.", "parameters": { "type": "object", "required": ["title", "objective"], "properties": { "title": {"type": "string"}, "objective": {"type": "string"}, "kind": {"type": "string", "enum": ["buying", "custom"], "default": "buying"}, "cadence": {"type": "string", "description": "Five-field cron expression, default every six hours."}, "constraints": {"type": "object", "description": "Plan-specific options such as message_tone, excluded_sellers, preferred_locations, max_unit_price, or custom instructions."}, "items": { "type": "array", "items": { "type": "object", "properties": { "item_name": {"type": "string"}, "desired_quantity": {"type": "integer", "minimum": 1}, "max_unit_price": {"type": "number"}, }, }, }, }, }, }, }, { "type": "function", "function": { "name": "list_continual_plans", "description": "List durable continual plans and their statuses.", "parameters": {"type": "object", "properties": {"include_inactive": {"type": "boolean", "default": True}}}, }, }, { "type": "function", "function": { "name": "get_continual_plan", "description": "Get one continual plan with checklist items, candidates, negotiations, and event history.", "parameters": {"type": "object", "required": ["plan_id"], "properties": {"plan_id": {"type": "string"}}}, }, }, { "type": "function", "function": { "name": "pause_continual_plan", "description": "Pause a continual plan so scheduled runs stop.", "parameters": {"type": "object", "required": ["plan_id"], "properties": {"plan_id": {"type": "string"}}}, }, }, { "type": "function", "function": { "name": "resume_continual_plan", "description": "Resume a paused or needs-input continual plan. It only becomes active when it has checklist items.", "parameters": {"type": "object", "required": ["plan_id"], "properties": {"plan_id": {"type": "string"}}}, }, }, { "type": "function", "function": { "name": "cancel_continual_plan", "description": "Cancel a continual plan.", "parameters": {"type": "object", "required": ["plan_id"], "properties": {"plan_id": {"type": "string"}}}, }, }, { "type": "function", "function": { "name": "delete_continual_plan", "description": "Delete a continual plan and all of its stored checklist items, candidates, negotiations, and event history.", "parameters": {"type": "object", "required": ["plan_id"], "properties": {"plan_id": {"type": "string"}}}, }, }, { "type": "function", "function": { "name": "run_continual_plan_now", "description": "Run one continual plan immediately and put the result in the Inbox.", "parameters": {"type": "object", "required": ["plan_id"], "properties": {"plan_id": {"type": "string"}}}, }, }, { "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)} @contextmanager def chat_image_scope(self, images: list[dict[str, Any]] | None): token = self._chat_images_var.set(self._normalize_chat_images(images)) try: yield finally: self._chat_images_var.reset(token) 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": result = await self.uex.delete(action.endpoint, action.payload, authenticated=True) else: result = await self.uex.post(action.endpoint, self._production_payload(action.endpoint, action.payload), authenticated=True) self._record_pending_action_result(action, "approved", result) return result 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}"} self._record_pending_action_result(action, "declined", {}) return { "declined": True, "pending_action": { "id": action.id, "label": action.label, "method": action.method, "endpoint": action.endpoint, "payload": self._display_payload(action.payload), "metadata": action.metadata or {}, }, } 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 _scmdb_schemas(cls) -> list[dict[str, Any]]: version_controls = { "version": {"type": "string", "description": "SCMDB game-data version, such as 4.7.2-live.11715810."}, "channel": {"type": "string", "enum": ["live", "ptu", "latest"], "default": "live"}, } return [ { "type": "function", "function": { "name": "list_scmdb_versions", "description": "List SCMDB mission-data versions. Use this when the user asks which Star Citizen game versions are available.", "parameters": { "type": "object", "properties": { "channel": {"type": "string", "enum": ["live", "ptu", "latest"]}, }, }, }, }, { "type": "function", "function": { "name": "search_scmdb_missions", "description": "Search SCMDB Star Citizen missions/contracts and return compact reward summaries: UEC, reputation, item, blueprint, and hauling rewards.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Text to search in title, debug name, description, faction, mission type, or reward names."}, "mission_type": {"type": "string", "description": "Mission type such as Hauling, Delivery, Bounty Hunter, Mercenary, Racing, Salvage, or Mining."}, "category": {"type": "string"}, "faction": {"type": "string"}, "system": {"type": "string", "description": "Star system such as Stanton or Pyro."}, "illegal": {"type": "boolean"}, "include_legacy": {"type": "boolean", "default": True}, "limit": {"type": "integer", "minimum": 1, "maximum": 25, "default": 10}, **version_controls, }, }, }, }, { "type": "function", "function": { "name": "get_scmdb_mission_rewards", "description": "Fetch detailed SCMDB rewards and requirements for one Star Citizen mission/contract by id, debug name, or title.", "parameters": { "type": "object", "properties": { "id": {"type": "string"}, "debug_name": {"type": "string"}, "title": {"type": "string"}, "include_legacy": {"type": "boolean", "default": True}, **version_controls, }, }, }, }, ] @classmethod def _scwiki_schemas(cls) -> list[dict[str, Any]]: return [ { "type": "function", "function": { "name": "search_scwiki_pages", "description": "Search Star Citizen Wiki pages on starcitizen.tools and return concise summaries for general game knowledge.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Page title or topic to search for."}, "limit": {"type": "integer", "minimum": 1, "maximum": 10, "default": 5}, }, }, }, }, { "type": "function", "function": { "name": "get_scwiki_page", "description": "Fetch one Star Citizen Wiki page summary by title or page id.", "parameters": { "type": "object", "properties": { "title": {"type": "string"}, "pageid": {"type": "integer"}, "chars": {"type": "integer", "minimum": 120, "maximum": 1200, "default": 700}, }, }, }, }, { "type": "function", "function": { "name": "search_scwiki_vehicles", "description": "Search Star Citizen Wiki structured vehicle data for ships and vehicles.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Ship or vehicle name to search for."}, "limit": {"type": "integer", "minimum": 1, "maximum": 10, "default": 5}, }, }, }, }, { "type": "function", "function": { "name": "get_scwiki_vehicle", "description": "Fetch one Star Citizen Wiki vehicle summary, including MSRP and in-game purchase locations when available.", "parameters": { "type": "object", "properties": { "slug": {"type": "string", "description": "Vehicle slug such as anvl-carrack."}, "query": {"type": "string", "description": "Vehicle name if the slug is not known."}, }, }, }, }, ] @classmethod def _wikelo_schemas(cls) -> list[dict[str, Any]]: return [ { "type": "function", "function": { "name": "search_wikelo_ship_projects", "description": "Search Wikelo ship projects and their required materials from wikelo-projects.com. Use this when the user asks for Wikelo ship requirements or build materials.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Ship or project name to search for, such as Polaris, Idris, Zeus, or Guardian."}, "limit": {"type": "integer", "minimum": 1, "maximum": 10, "default": 5}, }, }, }, }, { "type": "function", "function": { "name": "get_wikelo_ship_project", "description": "Fetch one Wikelo ship project with its required materials and contribution progress.", "parameters": { "type": "object", "properties": { "project_id": {"type": "string", "description": "Wikelo ship project id."}, "ship_name": {"type": "string", "description": "Ship or project name if the project id is not known."}, }, }, }, }, ] @classmethod def _cornerstone_schemas(cls) -> list[dict[str, Any]]: return [ { "type": "function", "function": { "name": "search_cornerstone_items", "description": "Search Cornerstone Universal Item Finder items. Use this to find exact item names and ids before asking where an item is sold.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Item name to search for."}, "sold_only": {"type": "boolean", "default": False, "description": "Only return items marked as sold in-game by Cornerstone."}, "limit": {"type": "integer", "minimum": 1, "maximum": 25, "default": 10}, }, }, }, }, { "type": "function", "function": { "name": "get_cornerstone_item_locations", "description": "Fetch where a Star Citizen item is sold using Cornerstone Universal Item Finder, including store/location, base price, and verified date.", "parameters": { "type": "object", "properties": { "id": {"type": "string", "description": "Cornerstone item id from search_cornerstone_items."}, "query": {"type": "string", "description": "Item name if id is not known."}, "location": {"type": "string", "description": "Optional local filter for system, planet, station, city, or shop name."}, "limit": {"type": "integer", "minimum": 1, "maximum": 50, "default": 20}, }, }, }, }, { "type": "function", "function": { "name": "get_cornerstone_item_media", "description": "Fetch Cornerstone item page media, especially image URLs that can be used when drafting UEX marketplace listings.", "parameters": { "type": "object", "properties": { "id": {"type": "string", "description": "Cornerstone item id from search_cornerstone_items."}, "query": {"type": "string", "description": "Item name if id is not known."}, "limit": {"type": "integer", "minimum": 1, "maximum": 10, "default": 5}, }, }, }, }, { "type": "function", "function": { "name": "draft_marketplace_listing_with_cornerstone_image", "description": "Draft a UEX marketplace listing and source the listing image from Cornerstone. The image is downloaded as base64 image_data and included in the pending action. Nothing is posted until user approval.", "parameters": { "type": "object", "required": ["item_query", "id_category", "operation", "type", "unit", "title", "description", "price", "currency", "language"], "properties": { "item_query": {"type": "string", "description": "Cornerstone item name to source an image from."}, "cornerstone_id": {"type": "string", "description": "Cornerstone item id, if already known."}, "id_item": {"type": "integer"}, "id_star_system": {"type": "integer"}, "id_terminal": {"type": "integer"}, "id_organization": {"type": "integer"}, "id_category": {"type": "integer"}, "operation": {"type": "string", "enum": ["buy", "sell", "rent", "trade"]}, "type": {"type": "string", "enum": ["item", "service", "contract"]}, "unit": {"type": "string"}, "title": {"type": "string"}, "description": {"type": "string"}, "price": {"type": "number"}, "currency": {"type": "string", "enum": ["UEC"]}, "language": {"type": "string", "default": "en_US"}, "location": {"type": "string"}, "source": {"type": "string", "enum": ["looted", "pledged", "purchased_in_game", "pirated", "gifted"]}, "availability": {"type": "string"}, "in_stock": {"type": "integer"}, "durability": {"type": "integer", "minimum": 0, "maximum": 100}, "video_url": {"type": "string"}, "image_data": {"type": "string", "description": "Base64 JPG or PNG image data for UEX upload."}, "use_attached_image": { "type": "boolean", "description": "When true, reuse an image pasted into the current chat as the listing image_data instead of sourcing from Cornerstone.", }, "attached_image_index": { "type": "integer", "minimum": 0, "description": "Zero-based pasted image index to reuse when use_attached_image is true.", }, "hours_expiration": {"type": "integer"}, "is_hidden": {"type": "integer", "enum": [0, 1]}, "is_tv_allowed": {"type": "integer", "enum": [0, 1]}, "is_production": {"type": "integer", "enum": [0, 1], "default": 1}, "require_image": {"type": "boolean", "default": False, "description": "Return an error instead of drafting if no Cornerstone JPG/PNG image can be sourced."}, }, }, }, }, ] @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 get_marketplace_trends( self, id_item: int | None = None, item_name: str | None = None, item_slug: str | None = None, id_category: int | None = None, currency: str | None = None, quality_tier: int | None = None, ) -> dict[str, Any]: response = await self.uex.get( "marketplace_trends", { "id_item": id_item, "item_name": item_name, "item_slug": item_slug, "id_category": id_category, "currency": currency, "quality_tier": quality_tier, }, ) trends = [ self._summarize_marketplace_trend(item) for item in self._as_list(response.get("data")) if isinstance(item, dict) ] return { "status": response.get("status"), "count": len(trends), "filters": { key: value for key, value in { "id_item": id_item, "item_name": item_name, "item_slug": item_slug, "id_category": id_category, "currency": currency, "quality_tier": quality_tier, }.items() if value is not None }, "trends": trends, } async def list_marketplace_negotiations( self, id: int | None = None, id_listing: int | None = None, hash: str | None = None, ) -> dict[str, Any]: response = await self.uex.get("marketplace_negotiations", {"id": id, "id_listing": id_listing, "hash": hash}, authenticated=True) negotiations = [ self._summarize_negotiation(item) for item in self._as_list(response.get("data")) if isinstance(item, dict) ] return {**response, "data": negotiations, "negotiations": negotiations} 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 list_local_negotiations( self, status: str = "all", unread_only: bool = False, search: str = "", limit: int = 10, ) -> dict[str, Any]: if self.negotiation_sync is None: return {"error": "Negotiation sync is not configured."} negotiations = self.negotiation_sync.list_negotiations( status=status, unread_only=unread_only, search=search, limit=limit, ) return {"count": len(negotiations), "negotiations": negotiations} async def get_local_negotiation(self, hash: str) -> dict[str, Any]: if self.negotiation_sync is None: return {"error": "Negotiation sync is not configured."} negotiation = self.negotiation_sync.get_negotiation(hash, mark_read=False) if not negotiation: return {"error": f"Negotiation not found: {hash}"} return {"negotiation": negotiation} async def search_local_negotiation_messages(self, query: str, limit: int = 8) -> dict[str, Any]: if self.negotiation_sync is None: return {"error": "Negotiation sync is not configured."} matches = self.negotiation_sync.search_messages(query, limit=limit) return {"count": len(matches), "matches": matches} async def draft_negotiation_message( self, message: str, hash: str | None = None, id_negotiation: int | None = None, id_listing: int | None = None, plan_id: str | None = None, plan_item_id: int | None = None, candidate_id: int | None = None, listing_slug: str | None = None, is_production: int = 1, ) -> dict[str, Any]: payload = {"message": message, "hash": hash, "id_negotiation": id_negotiation, "id_listing": id_listing, "is_production": is_production} metadata = { "plan_id": plan_id, "plan_item_id": plan_item_id, "candidate_id": candidate_id, "listing_id": id_listing, "listing_slug": listing_slug, "hash": hash, "id_negotiation": id_negotiation, } return self._pending("Send negotiation message", "marketplace_negotiations_messages", payload, metadata=metadata) async def draft_marketplace_listing(self, **payload: Any) -> dict[str, Any]: attached_image = self._attach_chat_image(payload) if attached_image.get("error"): return {"error": attached_image["error"]} return self._pending( "Post marketplace listing", "marketplace_advertise", payload, metadata=attached_image.get("metadata"), ) async def draft_negotiation_close( self, deal_closed: bool, hash: str | None = None, id_negotiation: int | None = None, deal_value: float | None = None, currency: str | None = None, clarity_rating: int | None = None, speed_rating: int | None = None, respect_rating: int | None = None, fairness_rating: int | None = None, comment: str | None = None, ) -> dict[str, Any]: payload = { "hash": hash, "id_negotiation": id_negotiation, "deal_closed": 1 if deal_closed else 0, "deal_value": deal_value, "currency": currency, "clarity_rating": clarity_rating, "speed_rating": speed_rating, "respect_rating": respect_rating, "fairness_rating": fairness_rating, "comment": comment, } metadata = { "hash": hash, "id_negotiation": id_negotiation, "kind": "negotiation_close", } return self._pending("Close negotiation", UEX_NEGOTIATION_CLOSE_ENDPOINT, payload, metadata=metadata) async def draft_negotiation_rating(self, **payload: Any) -> dict[str, Any]: return await self.draft_negotiation_close(**payload) async def draft_marketplace_listing_with_cornerstone_image( self, item_query: str, cornerstone_id: str | None = None, **payload: Any, ) -> dict[str, Any]: require_image = bool(payload.pop("require_image", False)) attached_image = self._attach_chat_image(payload) if attached_image.get("error"): return {"error": attached_image["error"]} item = await self._resolve_cornerstone_item(id=cornerstone_id, query=item_query) if not item: return {"error": "No Cornerstone item matched. Provide cornerstone_id or a more specific item_query."} page = await self.cornerstone.get_item_page(str(item["id"])) parsed = parse_cornerstone_item_page(page["html"], page["url"]) media = parsed.get("media") or [] image_result: dict[str, Any] | None = None image_error = "" for media_item in media: try: image_result = await self.cornerstone.get_image_data(media_item["url"]) break except Exception as exc: image_error = str(exc) if image_result and not payload.get("image_data"): payload["image_data"] = image_result["image_data"] elif require_image and not payload.get("image_data"): return { "error": "Cornerstone item matched, but no usable JPG/PNG image could be sourced.", "cornerstone": { "item": {"id": item.get("id"), "name": parsed.get("name") or item.get("name")}, "url": page["url"], "media": media, "image_error": image_error, }, } payload.setdefault("id_item", self._int_or_none(item.get("id"))) metadata = { "cornerstone_item_id": item.get("id"), "cornerstone_item_name": parsed.get("name") or item.get("name"), "cornerstone_url": page["url"], "cornerstone_image_url": image_result.get("url") if image_result else None, "cornerstone_image_content_type": image_result.get("content_type") if image_result else None, "cornerstone_image_size_bytes": image_result.get("size_bytes") if image_result else None, "cornerstone_image_status": "user_attached" if attached_image.get("metadata") else ("included" if image_result else "not_found"), "cornerstone_image_error": image_error or None, } if attached_image.get("metadata"): metadata.update(attached_image["metadata"]) return self._pending("Post marketplace listing with Cornerstone image", "marketplace_advertise", payload, metadata=metadata) 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 create_continual_plan( self, title: str, objective: str, kind: str = "buying", items: list[dict[str, Any]] | None = None, constraints: dict[str, Any] | None = None, cadence: str | None = None, ) -> dict[str, Any]: if self.plan_store is None: return {"error": "Continual plan store is not configured."} plan = self.plan_store.create_plan(title, kind=kind, objective=objective, items=items or [], constraints=constraints or {}, cadence=cadence) if self.scheduler is not None and plan.get("status") == "active": self.scheduler.schedule_plan(plan) plan = self.plan_store.get_plan(plan["id"]) or plan return {"plan": plan} async def list_continual_plans(self, include_inactive: bool = True) -> dict[str, Any]: if self.plan_store is None: return {"error": "Continual plan store is not configured."} return {"plans": self.plan_store.list_plans(include_inactive=include_inactive)} async def get_continual_plan(self, plan_id: str) -> dict[str, Any]: if self.plan_store is None: return {"error": "Continual plan store is not configured."} plan = self.plan_store.get_plan(plan_id) if not plan: return {"error": f"Plan not found: {plan_id}"} return {"plan": plan} async def pause_continual_plan(self, plan_id: str) -> dict[str, Any]: if self.plan_store is None: return {"error": "Continual plan store is not configured."} if self.scheduler is not None: self.scheduler.unschedule_plan(plan_id) return {"plan": self.plan_store.set_status(plan_id, "paused")} async def resume_continual_plan(self, plan_id: str) -> dict[str, Any]: if self.plan_store is None: return {"error": "Continual plan store is not configured."} plan = self.plan_store.get_plan(plan_id) if not plan: return {"error": f"Plan not found: {plan_id}"} next_status = "active" if plan.get("items") else "needs_input" plan = self.plan_store.set_status(plan_id, next_status) if self.scheduler is not None and plan and plan.get("status") == "active": self.scheduler.schedule_plan(plan) plan = self.plan_store.get_plan(plan_id) return {"plan": plan} async def cancel_continual_plan(self, plan_id: str) -> dict[str, Any]: if self.plan_store is None: return {"error": "Continual plan store is not configured."} if self.scheduler is not None: self.scheduler.unschedule_plan(plan_id) return {"plan": self.plan_store.set_status(plan_id, "canceled")} async def delete_continual_plan(self, plan_id: str) -> dict[str, Any]: if self.plan_store is None: return {"error": "Continual plan store is not configured."} plan = self.plan_store.get_plan(plan_id) if not plan: return {"error": f"Plan not found: {plan_id}"} if self.scheduler is not None: self.scheduler.unschedule_plan(plan_id) deleted = self.plan_store.delete_plan(plan_id) if not deleted: return {"error": f"Plan not found: {plan_id}"} return {"deleted": True, "plan_id": plan_id, "summary": f"Deleted plan {plan.get('title') or plan_id}."} async def run_continual_plan_now(self, plan_id: str) -> dict[str, Any]: if self.plan_runner is None: return {"error": "Continual plan runner is not configured."} return await self.plan_runner.run_plan(plan_id) 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} async def list_scmdb_versions(self, channel: str | None = None) -> dict[str, Any]: versions = await self.scmdb.list_versions() channel_filter = (channel or "").casefold().strip() if channel_filter in {"live", "ptu"}: versions = [ item for item in versions if f"-{channel_filter}." in str(item.get("version", "")).casefold() ] elif channel_filter not in {"", "latest"}: return {"error": "SCMDB channel must be live, ptu, or latest."} return { "source": self.scmdb.base_url, "count": len(versions), "versions": versions, "default_channel": "live", } async def search_scmdb_missions( self, query: str = "", mission_type: str | None = None, category: str | None = None, faction: str | None = None, system: str | None = None, illegal: bool | None = None, include_legacy: bool = True, limit: int = 10, version: str | None = None, channel: str = "live", ) -> dict[str, Any]: data = await self.scmdb.get_data(version=version, channel=channel) q = (query or "").casefold().strip() mission_type_filter = (mission_type or "").casefold().strip() category_filter = (category or "").casefold().strip() faction_filter = (faction or "").casefold().strip() system_filter = (system or "").casefold().strip() matched = [] for source, mission in self._scmdb_contracts(data, include_legacy=include_legacy): summary = self._summarize_scmdb_mission(data, mission, source=source) if mission_type_filter and mission_type_filter not in str(summary.get("mission_type") or "").casefold(): continue if category_filter and category_filter not in str(summary.get("category") or "").casefold(): continue if faction_filter and faction_filter not in str(summary.get("faction") or "").casefold(): continue if system_filter and system_filter not in " ".join(summary.get("systems") or []).casefold(): continue if illegal is not None and bool(summary.get("illegal")) != illegal: continue if q and q not in self._scmdb_search_text(data, mission, summary): continue matched.append(summary) limit = max(1, min(limit, 25)) return { "source": self.scmdb.base_url, "version": data.get("version"), "matched": len(matched), "returned": min(len(matched), limit), "truncated": len(matched) > limit, "missions": matched[:limit], } async def get_scmdb_mission_rewards( self, id: str | None = None, debug_name: str | None = None, title: str | None = None, include_legacy: bool = True, version: str | None = None, channel: str = "live", ) -> dict[str, Any]: if not any([id, debug_name, title]): return {"error": "Provide id, debug_name, or title."} data = await self.scmdb.get_data(version=version, channel=channel) exact = [] fuzzy = [] id_filter = (id or "").casefold().strip() debug_filter = (debug_name or "").casefold().strip() title_filter = (title or "").casefold().strip() for source, mission in self._scmdb_contracts(data, include_legacy=include_legacy): mission_id = str(mission.get("id") or "").casefold() mission_debug = str(mission.get("debugName") or "").casefold() mission_title = str(mission.get("title") or "").casefold() if id_filter and mission_id == id_filter: exact.append((source, mission)) elif debug_filter and mission_debug == debug_filter: exact.append((source, mission)) elif title_filter and mission_title == title_filter: exact.append((source, mission)) elif title_filter and title_filter in mission_title: fuzzy.append((source, mission)) elif debug_filter and debug_filter in mission_debug: fuzzy.append((source, mission)) candidates = exact or fuzzy if len(candidates) != 1: return { "source": self.scmdb.base_url, "version": data.get("version"), "matched": len(candidates), "error": "No SCMDB mission matched." if not candidates else "Multiple SCMDB missions matched; refine by id or debug_name.", "matches": [ self._summarize_scmdb_mission(data, mission, source=source) for source, mission in candidates[:10] ], } source, mission = candidates[0] return { "source": self.scmdb.base_url, "version": data.get("version"), "mission": self._summarize_scmdb_mission(data, mission, source=source, detailed=True), } async def search_scwiki_pages(self, query: str, limit: int = 5) -> dict[str, Any]: pages = await self.scwiki.search_pages(query, limit=limit) return {"source": self.scwiki.base_url, "query": query, "matched": len(pages), "pages": pages} async def get_scwiki_page( self, title: str | None = None, pageid: int | None = None, chars: int = 700, ) -> dict[str, Any]: page = await self.scwiki.get_page_summary(title=title, pageid=pageid, chars=chars) if not page: return {"error": "No Star Citizen Wiki page matched."} return {"source": self.scwiki.base_url, "page": page} async def search_scwiki_vehicles(self, query: str, limit: int = 5) -> dict[str, Any]: groups = await self.scwiki.search_verse(query) vehicles_group = next((item for item in groups if item.get("type") == "vehicles"), None) results = [ self._summarize_scwiki_vehicle_search(item) for item in (vehicles_group or {}).get("results", [])[: max(1, min(limit, 10))] if isinstance(item, dict) ] return {"source": self.scwiki.api_base_url, "query": query, "matched": len(results), "vehicles": results} async def get_scwiki_vehicle(self, slug: str | None = None, query: str | None = None) -> dict[str, Any]: resolved_slug = slug if not resolved_slug: if not query: return {"error": "Provide slug or query."} groups = await self.scwiki.search_verse(query) vehicles_group = next((item for item in groups if item.get("type") == "vehicles"), None) candidates = [ item for item in (vehicles_group or {}).get("results", []) if isinstance(item, dict) and item.get("api_url") ] if not candidates: return {"error": "No Star Citizen Wiki vehicle matched."} resolved_slug = str(candidates[0]["api_url"]).rstrip("/").rsplit("/", 1)[-1] vehicle = await self.scwiki.get_vehicle(resolved_slug) return {"source": self.scwiki.api_base_url, "vehicle": self._summarize_scwiki_vehicle(vehicle)} async def search_wikelo_ship_projects(self, query: str, limit: int = 5) -> dict[str, Any]: projects = await self.wikelo.list_ship_projects() q = (query or "").casefold().strip() matches = [] for project in projects: score = self._wikelo_ship_match_score(q, project) if q and score <= 0: continue matches.append((score, project)) matches.sort( key=lambda match: ( -match[0], str(match[1].get("ship_name") or "").casefold(), str(match[1].get("id") or ""), ) ) limit = max(1, min(limit, 10)) return { "source": f"{self.wikelo.base_url}/Ships", "query": query, "matched": len(matches), "projects": [self._summarize_wikelo_ship_project(item) for _, item in matches[:limit]], } async def get_wikelo_ship_project(self, project_id: str | None = None, ship_name: str | None = None) -> dict[str, Any]: projects = await self.wikelo.list_ship_projects() if project_id: for project in projects: if str(project.get("id") or "").strip() == str(project_id).strip(): return {"source": f"{self.wikelo.base_url}/Ships", "project": self._summarize_wikelo_ship_project(project, detailed=True)} return {"error": "No Wikelo ship project matched that id."} if not ship_name: return {"error": "Provide project_id or ship_name."} ranked = [ (self._wikelo_ship_match_score(ship_name.casefold().strip(), project), project) for project in projects ] ranked = [match for match in ranked if match[0] > 0] ranked.sort(key=lambda match: (-match[0], str(match[1].get("ship_name") or "").casefold())) if not ranked: return {"error": "No Wikelo ship project matched."} return {"source": f"{self.wikelo.base_url}/Ships", "project": self._summarize_wikelo_ship_project(ranked[0][1], detailed=True)} async def search_cornerstone_items( self, query: str = "", sold_only: bool = False, limit: int = 10, ) -> dict[str, Any]: items = await self.cornerstone.list_items() q = (query or "").casefold().strip() matches = [] for item in items: if sold_only and not item.get("sold"): continue score = self._cornerstone_match_score(q, str(item.get("name") or "")) if q and score <= 0: continue matches.append((score, item)) matches.sort(key=lambda match: (-match[0], str(match[1].get("name") or "").casefold())) limit = max(1, min(limit, 25)) compacted = [ { "id": item.get("id"), "name": item.get("name"), "sold": bool(item.get("sold")), "url": f"{self.cornerstone.base_url}/Search/{item.get('id')}", } for _, item in matches[:limit] ] return { "source": self.cornerstone.base_url, "matched": len(matches), "returned": len(compacted), "truncated": len(matches) > limit, "items": compacted, } async def get_cornerstone_item_locations( self, id: str | None = None, query: str | None = None, location: str | None = None, limit: int = 20, ) -> dict[str, Any]: item = await self._resolve_cornerstone_item(id=id, query=query) if not item: return {"error": "No Cornerstone item matched. Provide an id or a more specific query."} page = await self.cornerstone.get_item_page(str(item["id"])) parsed = parse_cornerstone_item_page(page["html"], page["url"]) locations = parsed.get("locations") or [] location_filter = (location or "").casefold().strip() if location_filter: locations = [ entry for entry in locations if location_filter in str(entry.get("location") or "").casefold() ] limit = max(1, min(limit, 50)) return { "source": self.cornerstone.base_url, "url": page["url"], "item": { "id": item.get("id"), "name": parsed.get("name") or item.get("name"), "sold": bool(item.get("sold")), "general": parsed.get("general") or {}, }, "matched_locations": len(locations), "returned": min(len(locations), limit), "truncated": len(locations) > limit, "locations": locations[:limit], } async def get_cornerstone_item_media( self, id: str | None = None, query: str | None = None, limit: int = 5, ) -> dict[str, Any]: item = await self._resolve_cornerstone_item(id=id, query=query) if not item: return {"error": "No Cornerstone item matched. Provide an id or a more specific query."} page = await self.cornerstone.get_item_page(str(item["id"])) parsed = parse_cornerstone_item_page(page["html"], page["url"]) media = parsed.get("media") or [] limit = max(1, min(limit, 10)) return { "source": self.cornerstone.base_url, "url": page["url"], "item": { "id": item.get("id"), "name": parsed.get("name") or item.get("name"), "sold": bool(item.get("sold")), "general": parsed.get("general") or {}, }, "returned": min(len(media), limit), "truncated": len(media) > limit, "media": media[:limit], } def _pending( self, label: str, endpoint: str, payload: dict[str, Any], method: str = "POST", metadata: dict[str, Any] | None = None, ) -> dict[str, Any]: action_id = str(uuid.uuid4()) payload = {key: value for key, value in payload.items() if value is not None} metadata = {key: value for key, value in (metadata or {}).items() if value is not None} payload = self._production_payload(endpoint, payload) self.pending_actions[action_id] = PendingAction(action_id, label, endpoint, payload, method, metadata) return { "pending_action": { "id": action_id, "label": label, "method": method, "endpoint": endpoint, "payload": self._display_payload(payload), "metadata": metadata, "approval_required": self.require_write_approval, } } @staticmethod def _display_payload(payload: dict[str, Any]) -> dict[str, Any]: display = dict(payload) image_data = display.get("image_data") if isinstance(image_data, str) and image_data: display["image_data"] = f"" return display def _attach_chat_image(self, payload: dict[str, Any]) -> dict[str, Any]: attached_index = payload.pop("attached_image_index", None) use_attached_image = bool(payload.pop("use_attached_image", False) or attached_index is not None) if payload.get("image_data") or not use_attached_image: return {} image = self._chat_image(attached_index or 0) if not image: return {"error": "No pasted chat image is available at the requested attached_image_index."} payload["image_data"] = image["image_data"] return { "metadata": { "attached_chat_image_name": image.get("name"), "attached_chat_image_content_type": image.get("content_type"), "attached_chat_image_index": attached_index or 0, "attached_chat_image_status": "included", } } def _chat_image(self, index: int) -> dict[str, Any] | None: images = self._chat_images_var.get() if 0 <= index < len(images): return images[index] return None @staticmethod def _normalize_chat_images(images: list[dict[str, Any]] | None) -> list[dict[str, Any]]: normalized: list[dict[str, Any]] = [] for image in images or []: if not isinstance(image, dict): continue image_data = str(image.get("image_data") or "").strip() if not image_data: continue normalized.append( { "name": str(image.get("name") or "").strip() or "pasted-image.png", "content_type": str(image.get("content_type") or "image/png").strip() or "image/png", "image_data": image_data, } ) return normalized @staticmethod def _int_or_none(value: Any) -> int | None: try: return int(value) except (TypeError, ValueError): return None def _record_pending_action_result(self, action: PendingAction, result_kind: str, result: dict[str, Any]) -> None: metadata = action.metadata or {} plan_id = metadata.get("plan_id") if not plan_id or self.plan_store is None: return message = f"{action.label} {result_kind} for continual plan." event_metadata = {"action_id": action.id, "endpoint": action.endpoint, "payload": action.payload, "result": result, **metadata} self.plan_store.add_event(plan_id, result_kind, message, event_metadata) if result_kind == "approved" and action.endpoint == "marketplace_negotiations_messages": self.plan_store.add_negotiation( plan_id, metadata.get("plan_item_id"), metadata.get("candidate_id"), {**metadata, "status": "approved"}, ) @staticmethod def _production_payload(endpoint: str, payload: dict[str, Any]) -> dict[str, Any]: if endpoint not in UEX_PRODUCTION_WRITE_RESOURCES: return payload next_payload = dict(payload) next_payload["is_production"] = 1 return next_payload @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 _scmdb_contracts(data: dict[str, Any], include_legacy: bool = True) -> list[tuple[str, dict[str, Any]]]: contracts = [ ("contracts", mission) for mission in data.get("contracts") or [] if isinstance(mission, dict) ] if include_legacy: contracts.extend( ("legacyContracts", mission) for mission in data.get("legacyContracts") or [] if isinstance(mission, dict) ) return contracts @classmethod def _summarize_scmdb_mission( cls, data: dict[str, Any], mission: dict[str, Any], source: str, detailed: bool = False, ) -> dict[str, Any]: summary: dict[str, Any] = { "id": mission.get("id"), "debug_name": mission.get("debugName"), "source": source, "title": mission.get("title"), "mission_type": mission.get("missionType"), "category": mission.get("category"), "faction": cls._scmdb_faction_name(data, mission.get("factionGuid")), "systems": mission.get("systems") or [], "illegal": bool(mission.get("illegal")), "can_be_shared": bool(mission.get("canBeShared")), "once_only": bool(mission.get("onceOnly")), "time_to_complete_minutes": mission.get("timeToComplete"), "max_players": mission.get("maxPlayersPerInstance"), "cooldown_minutes": mission.get("personalCooldownTime"), "min_standing": cls._scmdb_standing(mission.get("minStanding")), "max_standing": cls._scmdb_standing(mission.get("maxStanding")), "rewards": cls._scmdb_rewards(data, mission, detailed=detailed), } if detailed: summary.update( { "description": cls._compact_scalar(mission.get("description")), "locations": cls._scmdb_pool_names(data, "locationPools", mission.get("locations"), limit=20), "destinations": cls._scmdb_pool_names(data, "locationPools", mission.get("destinations"), limit=20), "prerequisites": cls._compact_scalar(mission.get("prerequisites")), "available_in_prison": bool(mission.get("availableInPrison")), "reaccept_after_abandoning": bool(mission.get("canReacceptAfterAbandoning")), "reaccept_after_failing": bool(mission.get("canReacceptAfterFailing")), "hide_in_mobiglas": bool(mission.get("hideInMobiGlas")), } ) return {key: value for key, value in summary.items() if value not in (None, "", [], {})} @classmethod def _scmdb_rewards(cls, data: dict[str, Any], mission: dict[str, Any], detailed: bool = False) -> dict[str, Any]: rewards: dict[str, Any] = { "uec": mission.get("rewardUEC"), "buy_in": mission.get("buyIn"), "dynamic_uec": mission.get("rewardIsDynamic"), "reputation": cls._scmdb_indexed_reputation(data, mission.get("factionRewardsIndex")), "failure_reputation": cls._scmdb_reputation_rewards(data, mission.get("factionRewards_fail")), "items": cls._scmdb_item_rewards(mission.get("itemRewards")), "blueprints": cls._scmdb_blueprint_rewards(data, mission.get("blueprintRewards"), detailed=detailed), "hauling": cls._scmdb_hauling_orders(data, mission.get("haulingOrders")), "partial_payouts": cls._scmdb_partial_payouts(data, mission.get("partialRewardPayoutIndex")), } return {key: value for key, value in rewards.items() if value not in (None, "", [], {})} @classmethod def _scmdb_indexed_reputation(cls, data: dict[str, Any], index: Any) -> list[dict[str, Any]]: if not isinstance(index, int): return [] pools = data.get("factionRewardsPools") or [] if index < 0 or index >= len(pools): return [] return cls._scmdb_reputation_rewards(data, pools[index]) @classmethod def _scmdb_reputation_rewards(cls, data: dict[str, Any], rewards: Any) -> list[dict[str, Any]]: if not isinstance(rewards, list): return [] result = [] for reward in rewards: if not isinstance(reward, dict): continue result.append( { "faction": cls._scmdb_faction_name(data, reward.get("factionGuid")), "scope": cls._scmdb_scope_name(data, reward.get("scopeGuid")), "amount": reward.get("amount"), } ) return [item for item in result if item.get("amount") not in (None, "")] @classmethod def _scmdb_item_rewards(cls, rewards: Any) -> list[dict[str, Any]]: if not isinstance(rewards, list): return [] return [ { "name": reward.get("name"), "amount": reward.get("amount"), } for reward in rewards if isinstance(reward, dict) ] @classmethod def _scmdb_blueprint_rewards(cls, data: dict[str, Any], rewards: Any, detailed: bool = False) -> list[dict[str, Any]]: if not isinstance(rewards, list): return [] result = [] for reward in rewards: if not isinstance(reward, dict): continue pool_id = reward.get("blueprintPool") pool = (data.get("blueprintPools") or {}).get(pool_id) or {} blueprints = pool.get("blueprints") or [] result.append( { "pool": pool.get("name") or reward.get("poolName"), "chance": reward.get("chance"), "trigger": reward.get("trigger"), "blueprints": [ cls._compact_scalar(item.get("name")) for item in blueprints[: 20 if detailed else 5] if isinstance(item, dict) and item.get("name") ], } ) return result @classmethod def _scmdb_hauling_orders(cls, data: dict[str, Any], orders: Any) -> list[dict[str, Any]]: if not isinstance(orders, list): return [] result = [] resources = data.get("resourcePools") or {} for order in orders: if not isinstance(order, dict): continue resource = resources.get(order.get("resource")) or {} result.append( { "resource": resource.get("name") or order.get("resource"), "min_scu": order.get("minSCU"), "max_scu": order.get("maxSCU"), "max_container_size_scu": order.get("maxContainerSize"), } ) return result @classmethod def _scmdb_partial_payouts(cls, data: dict[str, Any], index: Any) -> list[dict[str, Any]]: if not isinstance(index, int): return [] pools = data.get("partialRewardPayoutPools") or [] if index < 0 or index >= len(pools): return [] payouts = pools[index] if not isinstance(payouts, list): return [] return [ { "min_percent": payout.get("minPercentage"), "max_percent": payout.get("maxPercentage"), "currency_multiplier": payout.get("currencyRewardMultiplier"), "reputation_multipliers": payout.get("reputationMultipliers"), } for payout in payouts if isinstance(payout, dict) ] @staticmethod def _scmdb_faction_name(data: dict[str, Any], guid: Any) -> str | None: if not guid: return None faction = (data.get("factions") or {}).get(guid) if isinstance(faction, dict): return faction.get("name") or guid return str(guid) @staticmethod def _scmdb_scope_name(data: dict[str, Any], guid: Any) -> str | None: if not guid: return None scope = (data.get("scopes") or {}).get(guid) if isinstance(scope, dict): return scope.get("scopeName") or guid return str(guid) @staticmethod def _scmdb_standing(value: Any) -> dict[str, Any] | None: if not isinstance(value, dict): return None return { key: value.get(key) for key in ("name", "minReputation", "scopeName") if value.get(key) not in (None, "") } @staticmethod def _scmdb_pool_names(data: dict[str, Any], pool_key: str, keys: Any, limit: int = 10) -> list[str]: if not isinstance(keys, list): return [] pool = data.get(pool_key) or {} names = [] for key in keys[:limit]: item = pool.get(key) if isinstance(item, dict): names.append(str(item.get("name") or key)) else: names.append(str(key)) return names @classmethod def _scmdb_search_text(cls, data: dict[str, Any], mission: dict[str, Any], summary: dict[str, Any]) -> str: pieces = [ summary.get("title"), summary.get("debug_name"), summary.get("mission_type"), summary.get("category"), summary.get("faction"), mission.get("description"), " ".join(summary.get("systems") or []), ] rewards = summary.get("rewards") or {} pieces.append(str(rewards.get("uec") or "")) for item in rewards.get("reputation") or []: pieces.extend([item.get("faction"), item.get("scope"), item.get("amount")]) for item in rewards.get("items") or []: pieces.extend([item.get("name"), item.get("amount")]) for item in rewards.get("blueprints") or []: pieces.extend([item.get("pool"), " ".join(item.get("blueprints") or [])]) for item in rewards.get("hauling") or []: pieces.extend([item.get("resource"), item.get("min_scu"), item.get("max_scu")]) return " ".join(str(piece) for piece in pieces if piece not in (None, "")).casefold() async def _resolve_cornerstone_item(self, id: str | None = None, query: str | None = None) -> dict[str, Any] | None: items = await self.cornerstone.list_items() id_filter = (id or "").casefold().strip() if id_filter: for item in items: if str(item.get("id") or "").casefold() == id_filter: return item return {"id": id, "name": id, "sold": True} q = (query or "").casefold().strip() if not q: return None exact = [item for item in items if str(item.get("name") or "").casefold() == q] if exact: exact.sort(key=lambda item: not bool(item.get("sold"))) return exact[0] scored = [ (self._cornerstone_match_score(q, str(item.get("name") or "")), item) for item in items ] scored = [match for match in scored if match[0] > 0] if not scored: return None scored.sort(key=lambda match: (-match[0], not bool(match[1].get("sold")), str(match[1].get("name") or "").casefold())) return scored[0][1] @staticmethod def _cornerstone_match_score(query: str, name: str) -> int: if not query: return 1 normalized = name.casefold() if normalized == query: return 10000 if normalized.startswith(query): return 9000 - len(normalized) if query in normalized: return 8000 - normalized.index(query) tokens = [token for token in query.split() if token] if tokens and all(token in normalized for token in tokens): return 7000 - len(normalized) return 0 @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"), } @staticmethod def _summarize_marketplace_trend(trend: dict[str, Any]) -> dict[str, Any]: return { "id_item": trend.get("id_item"), "item_name": trend.get("item_name"), "item_slug": trend.get("item_slug"), "currency": trend.get("currency"), "sell": { "avg_price": trend.get("price_avg_sell"), "avg_price_month": trend.get("price_avg_month_sell"), "min_price": trend.get("price_min_sell"), "max_price": trend.get("price_max_sell"), "listings_count": trend.get("listings_count_sell"), }, "buy": { "avg_price": trend.get("price_avg_buy"), "avg_price_month": trend.get("price_avg_month_buy"), "min_price": trend.get("price_min_buy"), "max_price": trend.get("price_max_buy"), "listings_count": trend.get("listings_count_buy"), }, "total_listings_count": trend.get("total_listings_count"), "negotiations_count": trend.get("negotiations_count"), "negotiations_open": trend.get("negotiations_open"), "negotiations_success": trend.get("negotiations_success"), "link_prices": trend.get("link_prices"), "link_prices_history": trend.get("link_prices_history"), } @staticmethod def _summarize_scwiki_vehicle_search(vehicle: dict[str, Any]) -> dict[str, Any]: return { "name": vehicle.get("name"), "class_name": vehicle.get("class_name"), "career": vehicle.get("extra_label"), "api_url": vehicle.get("api_url"), "web_url": vehicle.get("web_url"), } @staticmethod def _summarize_scwiki_vehicle(vehicle: dict[str, Any]) -> dict[str, Any]: purchases = [] for entry in ((vehicle.get("uex_prices") or {}).get("purchase") or []): if not isinstance(entry, dict): continue location = entry.get("starmap_location") or {} purchases.append( { "price_buy": entry.get("price_buy"), "terminal_name": entry.get("terminal_name"), "location": location.get("name"), "parent_location": location.get("parent_name"), "star_system": location.get("star_system_name"), "game_version": entry.get("game_version"), "date_updated": entry.get("date_updated"), "uex_link": entry.get("uex_link"), } ) return { "name": vehicle.get("name") or vehicle.get("game_name"), "game_name": vehicle.get("game_name"), "slug": vehicle.get("slug"), "manufacturer": (vehicle.get("manufacturer") or {}).get("name"), "career": vehicle.get("career"), "role": vehicle.get("role"), "size_class": vehicle.get("size_class"), "cargo_capacity": vehicle.get("cargo_capacity"), "crew": vehicle.get("crew"), "msrp": vehicle.get("msrp"), "pledge_url": vehicle.get("pledge_url"), "purchase_locations": purchases, "description": ((vehicle.get("description") or {}).get("en_EN") or (vehicle.get("game_description") or {}).get("en_EN")), "web_url": vehicle.get("web_url"), "updated_at": vehicle.get("updated_at"), "version": vehicle.get("version"), } @staticmethod def _wikelo_ship_match_score(query: str, project: dict[str, Any]) -> int: if not query: return 1 ship_name = str(project.get("ship_name") or "").casefold() description = str(project.get("description") or "").casefold() materials = " ".join( str(item.get("material_name") or "").casefold() for item in (project.get("required_materials") or []) if isinstance(item, dict) ) haystack = " ".join(part for part in [ship_name, description, materials] if part) if ship_name == query: return 10000 if query in ship_name: return 9000 - ship_name.index(query) if query in description: return 7000 - description.index(query) if query in materials: return 5000 - materials.index(query) tokens = [token for token in query.split() if token] if tokens and all(token in haystack for token in tokens): return 3000 - len(haystack) return 0 @classmethod def _summarize_wikelo_ship_project(cls, project: dict[str, Any], detailed: bool = False) -> dict[str, Any]: materials = [] for item in (project.get("required_materials") or []): if not isinstance(item, dict): continue quantity_needed = item.get("quantity_needed") quantity_collected = item.get("quantity_collected") materials.append( { "material_name": item.get("material_name"), "quantity_needed": int(quantity_needed) if isinstance(quantity_needed, (int, float)) and float(quantity_needed).is_integer() else quantity_needed, "quantity_collected": int(quantity_collected) if isinstance(quantity_collected, (int, float)) and float(quantity_collected).is_integer() else quantity_collected, } ) summary = { "id": project.get("id"), "ship_name": project.get("ship_name"), "description": project.get("description"), "status": project.get("status"), "privacy": project.get("privacy"), "owner_name": project.get("owner_name"), "org_name": project.get("org_name"), "home_port": project.get("home_port"), "ship_image": project.get("ship_image"), "materials_count": len(materials), "required_materials": materials if detailed else materials[:12], "source_url": f"https://wikelo-projects.com/Ships", } return {key: value for key, value in summary.items() if value not in (None, "", [], {})} @classmethod def _summarize_negotiation(cls, negotiation: dict[str, Any]) -> dict[str, Any]: summary = cls._project_item(negotiation, mode="summary") state = cls._negotiation_state(negotiation) summary.update( { "state": state["state"], "is_open": state["is_open"], "state_reason": state["reason"], } ) for key in ("hash", "id_listing", "id_user", "id_user_seller", "id_user_buyer", "date_closed"): if key in negotiation and key not in summary: summary[key] = negotiation.get(key) return summary @staticmethod def _negotiation_state(negotiation: dict[str, Any]) -> dict[str, Any]: closed_flags = [ "is_closed", "closed", "is_cancelled", "is_canceled", "is_archived", "marked_closed", ] for key in closed_flags: value = negotiation.get(key) if value in (True, 1, "1", "true", "True", "yes", "closed"): return {"state": "closed", "is_open": False, "reason": f"{key} is set"} closed_dates = [ "date_closed", "date_completed", "date_cancelled", "date_canceled", "closed_at", "completed_at", "cancelled_at", "canceled_at", ] for key in closed_dates: value = negotiation.get(key) if value not in (None, "", 0, "0", False): return {"state": "closed", "is_open": False, "reason": f"{key} is populated"} status = str(negotiation.get("status") or negotiation.get("state") or "").casefold() if status in {"closed", "cancelled", "canceled", "completed", "declined", "accepted", "rejected"}: return {"state": "closed", "is_open": False, "reason": f"status is {status}"} if status in {"open", "active", "pending", "new"}: return {"state": "open", "is_open": True, "reason": f"status is {status}"} return {"state": "open", "is_open": True, "reason": "no closed flag, closed date, or closed status was present"}