Build ebay agent

.env.example

@@ -0,0 +1,10 @@
+ # OpenAI configuration
+ OPENAI_API_KEY=
+ OPENAI_API_BASE=
+
+ # MCP server configuration
+ MCP_ENDPOINT=
+ MCP_API_KEY=
+
+ # Optional overrides
+ LOG_LEVEL=INFO

.gitignore

@@ -0,0 +1,6 @@
+.env
+.venv
+__pycache__
+*.log
+*.csv
+*.egg-info

AGENTS.md

@@ -0,0 +1,62 @@
+# eBay Inventory & Listing Automation Tool — Agent Context
+
+This file gives LLM-based agents the minimum context needed to navigate and extend the project safely.
+
+## Mission
+
+Automate the pipeline from raw photo dumps to live eBay listings:
+
+1. **Organize mode (`--organize`)**: GPT-vision batches (≤50 images) identify handwritten SKU labels, group images, and rename the label shot to `SKU.<ext>`.
+2. **Inventory mode (default argument path)**: For each organized folder, GPT summarizes metadata while the MCP server provides sold-price comps; results are written to `listings_YYYYMMDD_HHMMSS.csv`.
+3. **Publish mode (`--upload`)**: Reads a CSV, uploads images via MCP, creates listings (draft or live), publishes when requested, and persists the returned `mcp_listing_id` back into the CSV.
+
+## Architecture Cheat Sheet
+
+| Module | Purpose |
+| --- | --- |
+| `app.py` | Typer CLI entrypoint. Determines mode, sets up logging/config, and runs async workflows. |
+| `core/env.py` | Loads `.env`, validates mode-specific requirements (`OPENAI_API_KEY`, `MCP_ENDPOINT`, etc.), exposes `AppConfig`. |
+| `core/models.py` | Pydantic models for folders, image assignments, CSV rows (see `CSV_HEADERS`), MCP payloads, and pricing responses. |
+| `core/csv_io.py` | Timestamped CSV writer, CSV reader, and overwrite helper for publish mode. |
+| `gpt/organize.py` | Handles batching, image base64 encoding, prompt construction, and JSON parsing for folder assignments. |
+| `gpt/analyze.py` | Builds listing metadata via GPT, appends description suffix `\n\nItem: <sku>`, and combines MCP sold-price data. |
+| `mcp/client.py` | Async `httpx` client with retry/backoff wrapping MCP tools (`upload_image`, `create_listing`, `publish_listing`, `search_sold`). |
+| `mcp/listings.py` | Orchestrates CSV-driven listing creation/publishing, resolves local image paths, enforces concurrency limits, and rewrites CSV rows. |
+| `utils/files.py` | Filesystem helpers to detect images, ensure folders, move/rename files after organize mode, and collect SKU folders. |
+| `utils/concurrency.py` | Semaphore-bound gather helpers that preserve ordering for deterministic CSV updates. |
+
+## External Services
+
+- **OpenAI Responses API** (`gpt-5`): used for both organization (vision) and inventory analysis (vision + text). Always request JSON, respect batch limits, and send base64 images.
+- **eBay MCP server** (`MCP_ENDPOINT`, `MCP_API_KEY`): provides search/comps, image upload, listing creation, and publishing. All eBay actions go through MCP tool invocations—never call eBay REST APIs directly.
+
+## Environment & Configuration
+
+- Copy `.env.example` → `.env`. Variables per mode:
+  - Organize: `OPENAI_API_KEY`
+  - Inventory: `OPENAI_API_KEY`, `MCP_ENDPOINT`, `MCP_API_KEY`
+  - Publish: `MCP_ENDPOINT`, `MCP_API_KEY`
+- Optional: `OPENAI_API_BASE`, `LOG_LEVEL`, `CONCURRENCY_LIMIT` (defaults to 5; controls folder/image parallelism).
+- Config access pattern: `config = get_config_for_mode("<mode>")`, then pass through to downstream modules.
+
+## Operational Notes
+
+- **Concurrency:** Use helpers in `utils.concurrency`; don’t spawn unbounded tasks. Inventory folders and MCP uploads must respect `config.concurrency_limit`.
+- **HTTP layer:** Always use `httpx.AsyncClient` with retries (already implemented in `mcp/client.py`). No `requests`.
+- **Filesystem:** `utils.files.move_assignments` renames the label image to match folder name and skips missing files gracefully.
+- **CSV schema:** `core/models.CSV_HEADERS` defines 24 columns. Descriptions must end with `Item: <folder>`. `mcp_listing_id` starts blank and is set during publish flow.
+- **Logging:** `core/logging.setup_logging` provides a uniform format. Use `get_logger(__name__)`.
+- **Error handling:** Missing env vars raise `ConfigError`. CLI catches and prints messages via Typer before exiting with status 1.
+
+## How to Extend Safely
+
+- Keep functions small/pure where possible; inject side effects (clients, paths, configs) for easier testing.
+- When adding new MCP interactions, update `mcp/client.py` so retries/backoff stay consistent.
+- Any new GPT flow should mirror existing JSON-only prompting style to keep parsing predictable.
+- Update both `README.md` (user-facing) and this file (agent-facing) when workflows change.
+
+## Learnings
+
+Record notable clarifications or user-specific expectations below.
+
+- _None yet_

README.md

@@ -1,2 +1,54 @@
-# ebay-
-AI driven ebay listing creator
+# Inventory & Listing Automation Tool
+
+AI-assisted workflow for organizing raw photo drops, generating inventory CSVs, and creating/publishing eBay listings through the MCP server.
+
+## Prerequisites
+
+- Python 3.11+
+- Access to the eBay MCP server (`MCP_ENDPOINT`, `MCP_API_KEY`)
+- OpenAI API key with GPT-5 access
+
+## Setup
+
+```bash
+git clone <repo>
+cd <repo>
+python -m venv .venv
+source .venv/bin/activate
+pip install -e .
+cp .env.example .env  # fill in secrets
+```
+
+## CLI Overview
+
+```
+python app.py --organize /path/to/raw_images
+python app.py /path/to/organized_items
+python app.py --upload /path/to/listings.csv [--draft|--publish]
+```
+
+- `--organize`: groups loose photos into SKU folders (max 50 images per GPT batch) and renames label shots to `SKU.jpg`.
+- `inventory` (default): scans organized folders, calls GPT for metadata, queries MCP for sold comps, and writes `listings_YYYYMMDD_HHMMSS.csv` in the provided root. Prints absolute CSV path on success.
+- `--upload`: reads a CSV, uploads images via MCP, and creates/publishes listings.
+  - Omit flags to create + publish in one pass.
+  - `--draft`: create drafts only.
+  - `--publish`: publish previously-created drafts (`mcp_listing_id` required per row).
+
+## Environment Variables
+
+See `.env.example`. Required per mode:
+
+| Mode      | Variables                               |
+|-----------|------------------------------------------|
+| Organize  | `OPENAI_API_KEY`                         |
+| Inventory | `OPENAI_API_KEY`, `MCP_ENDPOINT`, `MCP_API_KEY` |
+| Publish   | `MCP_ENDPOINT`, `MCP_API_KEY`            |
+
+Optional: `OPENAI_API_BASE`, `LOG_LEVEL`, `CONCURRENCY_LIMIT`.
+
+## Architecture Highlights
+
+- Async `httpx` client with retry/backoff for MCP interactions.
+- Pydantic models for CSV rows, price data, and listing payloads.
+- Structured concurrency helpers for bounded parallelism.
+- GPT-driven vision prompts for organization and metadata extraction (OpenAI Responses API).

app.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Optional
+
+import typer
+
+from core.csv_io import write_inventory_csv
+from core.env import ConfigError, get_config_for_mode
+from core.logging import setup_logging
+from gpt.analyze import analyze_inventory
+from gpt.organize import organize_images
+from mcp.listings import process_uploads
+from utils.files import move_assignments
+
+app = typer.Typer(add_completion=False, help="Inventory and listing automation tool.")
+
+
+def _run_async(coro):
+    return asyncio.run(coro)
+
+
+@app.command()
+def main(
+    items_path: Optional[Path] = typer.Argument(
+        None,
+        help="Path to organized item folders (inventory mode).",
+    ),
+    organize: Optional[Path] = typer.Option(
+        None,
+        "--organize",
+        exists=True,
+        dir_okay=True,
+        file_okay=False,
+        help="Path to raw images root.",
+    ),
+    upload: Optional[Path] = typer.Option(
+        None,
+        "--upload",
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        help="Path to inventory CSV for publish mode.",
+    ),
+    draft: bool = typer.Option(False, "--draft", help="Create draft listings only."),
+    publish: bool = typer.Option(False, "--publish", help="Publish existing drafts."),
+) -> None:
+    try:
+        if organize:
+            _run_organize_mode(organize)
+        elif upload:
+            _run_publish_mode(upload, draft, publish)
+        elif items_path:
+            _run_inventory_mode(items_path)
+        else:
+            raise typer.BadParameter("Provide --organize, --upload, or an items path.")
+    except ConfigError as exc:
+        typer.echo(str(exc), err=True)
+        raise typer.Exit(code=1) from exc
+
+
+def _run_organize_mode(path: Path) -> None:
+    if not path.exists():
+        raise typer.BadParameter(f"{path} not found.")
+    config = get_config_for_mode("organize")
+    setup_logging(config.log_level)
+    assignments = _run_async(organize_images(path, config))
+    move_assignments(assignments, path)
+    typer.echo(f"Organized {len(assignments)} images under {path}")
+
+
+def _run_inventory_mode(path: Path) -> None:
+    if not path.exists():
+        raise typer.BadParameter(f"{path} not found.")
+    config = get_config_for_mode("inventory")
+    setup_logging(config.log_level)
+    rows = _run_async(analyze_inventory(path, config))
+    csv_path = write_inventory_csv(rows, path)
+    typer.echo(f"Created CSV at {csv_path.resolve()}")
+
+
+def _run_publish_mode(csv_path: Path, draft: bool, publish: bool) -> None:
+    if draft and publish:
+        raise typer.BadParameter("Use either --draft or --publish, not both.")
+    config = get_config_for_mode("publish")
+    setup_logging(config.log_level)
+    updated_rows = _run_async(process_uploads(csv_path, draft_only=draft, publish_only=publish, config=config))
+    typer.echo(f"Updated {csv_path} for {len(updated_rows)} rows")
+
+
+def main_cli() -> None:
+    app()
+
+
+if __name__ == "__main__":
+    main_cli()

core/csv_io.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import csv
+from datetime import datetime
+from pathlib import Path
+from typing import Sequence
+
+from core.models import CSV_HEADERS, InventoryRow
+
+
+def _parse_float_list(value: str) -> list[float]:
+    if not value:
+        return []
+    return [float(part) for part in value.split(";") if part]
+
+
+def _parse_str_list(value: str) -> list[str]:
+    if not value:
+        return []
+    return [part for part in value.split(";") if part]
+
+
+def read_inventory_csv(csv_path: Path) -> list[InventoryRow]:
+    with csv_path.open("r", newline="", encoding="utf-8") as handle:
+        reader = csv.DictReader(handle)
+        rows: list[InventoryRow] = []
+        for row in reader:
+            rows.append(
+                InventoryRow(
+                    item_number=row.get("item_number", ""),
+                    sku=row.get("sku", ""),
+                    folder=row.get("folder", ""),
+                    title=row.get("title", ""),
+                    description=row.get("description", ""),
+                    condition=row.get("condition", "USED"),
+                    condition_note=row.get("condition_note", "") or "",
+                    recommended_price=float(row.get("recommended_price") or 0),
+                    price_rationale=row.get("price_rationale", "") or "",
+                    suggested_category_id=row.get("suggested_category_id", "") or "",
+                    suggested_category_path=row.get("suggested_category_path", "") or "",
+                    ebay_search_query=row.get("ebay_search_query", "") or "",
+                    last_sold_prices=_parse_float_list(row.get("last_sold_prices", "")),
+                    price_source=row.get("price_source", "") or "",
+                    highest_sold_price=(
+                        float(row["highest_sold_price"])
+                        if row.get("highest_sold_price")
+                        else None
+                    ),
+                    image_urls=_parse_str_list(row.get("image_urls", "")),
+                    notes_path=row.get("notes_path", "") or "",
+                    listing_format=row.get("listing_format", "") or "",
+                    marketplace_id=row.get("marketplace_id", "") or "",
+                    category_id=row.get("category_id", "") or "",
+                    quantity=int(row.get("quantity") or 1),
+                    listing_duration=row.get("listing_duration", "") or "",
+                    best_offer_enabled=(
+                        row.get("best_offer_enabled", "true").lower() == "true"
+                    ),
+                    mcp_listing_id=row.get("mcp_listing_id", "") or "",
+                )
+            )
+    return rows
+
+
+def _timestamped_filename(root: Path) -> Path:
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    return root / f"listings_{timestamp}.csv"
+
+
+def write_inventory_csv(rows: Sequence[InventoryRow], root_path: Path) -> Path:
+    csv_path = _timestamped_filename(root_path)
+    root_path.mkdir(parents=True, exist_ok=True)
+
+    with csv_path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=CSV_HEADERS)
+        writer.writeheader()
+        for row in rows:
+            writer.writerow(row.csv_row())
+
+    return csv_path
+
+
+def overwrite_inventory_csv(csv_path: Path, rows: Sequence[InventoryRow]) -> None:
+    with csv_path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=CSV_HEADERS)
+        writer.writeheader()
+        for row in rows:
+            writer.writerow(row.csv_row())

core/env.py

@@ -0,0 +1,60 @@
+ from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Iterable, Sequence
+
+from dotenv import load_dotenv
+from pydantic import BaseModel, Field, HttpUrl, PositiveInt, ValidationError
+
+
+class ConfigError(RuntimeError):
+    """Raised when required configuration is missing."""
+
+
+class AppConfig(BaseModel):
+    openai_api_key: str | None = Field(default=None, alias="OPENAI_API_KEY")
+    openai_api_base: HttpUrl | None = Field(default=None, alias="OPENAI_API_BASE")
+    mcp_endpoint: HttpUrl | None = Field(default=None, alias="MCP_ENDPOINT")
+    mcp_api_key: str | None = Field(default=None, alias="MCP_API_KEY")
+    log_level: str = Field(default="INFO", alias="LOG_LEVEL")
+    concurrency_limit: PositiveInt = Field(default=5, alias="CONCURRENCY_LIMIT")
+
+    model_config = {"populate_by_name": True}
+
+    def require(self, fields: Sequence[str]) -> None:
+        missing = [field for field in fields if not getattr(self, field)]
+        if missing:
+            raise ConfigError(
+                f"Missing required environment variables: {', '.join(missing)}"
+            )
+
+
+def _load_env_files() -> None:
+    env_path = Path(".env")
+    load_dotenv(dotenv_path=env_path if env_path.exists() else None, override=False)
+
+
+REQUIRED_BY_MODE = {
+    "organize": ("openai_api_key",),
+    "inventory": ("openai_api_key", "mcp_endpoint", "mcp_api_key"),
+    "publish": ("mcp_endpoint", "mcp_api_key"),
+}
+
+
+@lru_cache(maxsize=1)
+def get_config(required: Iterable[str] | None = None) -> AppConfig:
+    _load_env_files()
+    try:
+        config = AppConfig()  # type: ignore[call-arg]
+    except ValidationError as exc:
+        raise ConfigError(str(exc)) from exc
+
+    if required:
+        config.require(tuple(required))
+    return config
+
+
+def get_config_for_mode(mode: str) -> AppConfig:
+    requirements = REQUIRED_BY_MODE.get(mode, ())
+    return get_config(requirements)