LambdaBankingConglomerate/TraderAI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity
Files
main
TraderAI/traderai/tools.py
T
HRiggs 8fac3d2bae feat: chat
2026-06-09 11:24:15 -04:00

2854 lines
130 KiB
Python
Raw Permalink Blame History

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"<base64 image data redacted; {len(image_data)} characters>"
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"}
Reference in New Issue View Git Blame Copy Permalink