diff --git a/backend/app/dependencies.py b/backend/app/dependencies.py
index 2f8b597..9802209 100644
--- a/backend/app/dependencies.py
+++ b/backend/app/dependencies.py
@@ -53,3 +53,33 @@ async def get_current_user(
 
 
 CurrentUser = Annotated[User, Depends(get_current_user)]
+
+oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False)
+
+
+async def get_optional_user(
+    token: Annotated[str | None, Depends(oauth2_scheme_optional)],
+    db: Annotated[AsyncSession, Depends(get_db)],
+) -> User | None:
+    """Resolve the JWT bearer token to a User if present and valid, otherwise None."""
+    if not token:
+        return None
+    try:
+        payload = jwt.decode(
+            token, settings.public_pem, algorithms=[settings.ALGORITHM]
+        )
+        user_id: str | None = payload.get("sub")
+        if user_id is None:
+            return None
+        user_id_int = int(user_id)
+    except (JWTError, ValueError, TypeError):
+        return None
+
+    result = await db.execute(select(User).where(User.id == user_id_int))
+    user = result.scalar_one_or_none()
+    if user is None or not user.is_active:
+        return None
+    return user
+
+
+OptionalUser = Annotated[User | None, Depends(get_optional_user)]
diff --git a/backend/app/main.py b/backend/app/main.py
index 4ba9f69..699ffd9 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -9,7 +9,7 @@ from app.config import settings
 from app.routers import (
     auth, users, worlds, assets, avatar, world_content, internal,
     presence, groups, social_friends, profiles, chat_dm, chat_reports,
-    chat_world, wallet, payments,
+    chat_world, wallet, payments, marketplace,
 )
 from app.gateway import ws
 
@@ -60,6 +60,7 @@ app.include_router(chat_world.router, prefix="/api/chat", tags=["chat-world"])
 app.include_router(chat_world.router, prefix="/v1/chat", tags=["chat-world"])
 app.include_router(wallet.router, prefix="/api/wallet", tags=["wallet"])
 app.include_router(payments.router, prefix="/api/payments", tags=["payments"])
+app.include_router(marketplace.router, prefix="/api/marketplace", tags=["marketplace"])
 app.include_router(ws.router, prefix="/api", tags=["websocket"])
 app.include_router(ws.router, prefix="/v1", tags=["websocket"])
 
diff --git a/backend/app/routers/marketplace.py b/backend/app/routers/marketplace.py
new file mode 100644
index 0000000..78b80f6
--- /dev/null
+++ b/backend/app/routers/marketplace.py
@@ -0,0 +1,328 @@
+"""Marketplace listings API router.
+
+docs/planning/economy.md US-3 (task f51c7d95e237).
+"""
+
+from datetime import datetime
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel, Field
+from sqlalchemy import func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database import get_db
+from app.dependencies import CurrentUser, OptionalUser
+from app.models.marketplace import (
+    CatalogItem,
+    ItemType,
+    ListingStatus,
+    MarketplaceListing,
+)
+from app.models.user import User
+
+router = APIRouter()
+
+DbDep = Annotated[AsyncSession, Depends(get_db)]
+
+
+# ── Schemas ──
+
+class CatalogItemCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=100)
+    description: str | None = Field(default=None, max_length=2000)
+    item_type: ItemType
+    asset_id: int | None = None
+    metadata_json: dict[str, Any] | None = None
+
+
+class CatalogItemResponse(BaseModel):
+    id: int
+    owner_id: int
+    asset_id: int | None
+    item_type: ItemType
+    name: str
+    description: str | None
+    metadata_json: dict[str, Any] | None
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class ListingCreate(BaseModel):
+    catalog_item_id: int
+    price: int = Field(..., gt=0, description="Price in whole Credits (must be positive)")
+
+
+class ListingUpdate(BaseModel):
+    price: int | None = Field(default=None, gt=0)
+    status: ListingStatus | None = None
+
+
+class ListingResponse(BaseModel):
+    id: int
+    catalog_item_id: int
+    seller_id: int
+    price: int
+    status: ListingStatus
+    created_at: datetime
+    updated_at: datetime
+    catalog_item: CatalogItemResponse | None = None
+    seller_username: str | None = None
+
+    model_config = {"from_attributes": True}
+
+
+class ListingListResponse(BaseModel):
+    listings: list[ListingResponse]
+    total: int
+    limit: int
+    offset: int
+
+
+# ── Catalog Items ──
+
+@router.post("/items", response_model=CatalogItemResponse, status_code=status.HTTP_201_CREATED)
+async def create_catalog_item(
+    payload: CatalogItemCreate,
+    current_user: CurrentUser,
+    db: DbDep,
+):
+    """Create a new catalog item owned by the authenticated user."""
+    item = CatalogItem(
+        owner_id=current_user.id,
+        asset_id=payload.asset_id,
+        item_type=payload.item_type,
+        name=payload.name.strip(),
+        description=payload.description.strip() if payload.description else None,
+        metadata_json=payload.metadata_json,
+    )
+    db.add(item)
+    await db.commit()
+    await db.refresh(item)
+    return item
+
+
+@router.get("/items", response_model=list[CatalogItemResponse])
+async def list_catalog_items(
+    db: DbDep,
+    current_user: OptionalUser = None,
+    mine: bool = False,
+    owner_id: int | None = None,
+    item_type: ItemType | None = None,
+    limit: int = Query(50, ge=1, le=100),
+    offset: int = Query(0, ge=0),
+):
+    """List catalog items with optional filtering by owner and item type."""
+    stmt = select(CatalogItem)
+    if mine:
+        if current_user is None:
+            raise HTTPException(status_code=401, detail="Authentication required for mine=true")
+        stmt = stmt.where(CatalogItem.owner_id == current_user.id)
+    elif owner_id is not None:
+        stmt = stmt.where(CatalogItem.owner_id == owner_id)
+
+    if item_type is not None:
+        stmt = stmt.where(CatalogItem.item_type == item_type)
+
+    stmt = stmt.order_by(CatalogItem.created_at.desc(), CatalogItem.id.desc()).offset(offset).limit(limit)
+    res = await db.execute(stmt)
+    return list(res.scalars().all())
+
+
+# ── Listings ──
+
+@router.post("/listings", response_model=ListingResponse, status_code=status.HTTP_201_CREATED)
+async def create_listing(
+    payload: ListingCreate,
+    current_user: CurrentUser,
+    db: DbDep,
+):
+    """Create a new marketplace listing for an owned catalog item."""
+    item = await db.get(CatalogItem, payload.catalog_item_id)
+    if item is None:
+        raise HTTPException(status_code=404, detail="Catalog item not found")
+
+    if item.owner_id != current_user.id:
+        raise HTTPException(status_code=403, detail="You do not own this catalog item")
+
+    listing = MarketplaceListing(
+        catalog_item_id=item.id,
+        seller_id=current_user.id,
+        price=payload.price,
+        status=ListingStatus.active,
+    )
+    db.add(listing)
+    await db.commit()
+    await db.refresh(listing)
+
+    return ListingResponse(
+        id=listing.id,
+        catalog_item_id=listing.catalog_item_id,
+        seller_id=listing.seller_id,
+        price=listing.price,
+        status=listing.status,
+        created_at=listing.created_at,
+        updated_at=listing.updated_at,
+        catalog_item=CatalogItemResponse.model_validate(item),
+        seller_username=current_user.username,
+    )
+
+
+@router.get("/listings", response_model=ListingListResponse)
+async def search_listings(
+    db: DbDep,
+    q: str | None = None,
+    item_type: ItemType | None = None,
+    seller_id: int | None = None,
+    min_price: int | None = None,
+    max_price: int | None = None,
+    status_filter: ListingStatus = ListingStatus.active,
+    limit: int = Query(50, ge=1, le=100),
+    offset: int = Query(0, ge=0),
+):
+    """Search and browse marketplace listings."""
+    stmt = (
+        select(MarketplaceListing, CatalogItem, User.username)
+        .join(CatalogItem, MarketplaceListing.catalog_item_id == CatalogItem.id)
+        .join(User, MarketplaceListing.seller_id == User.id)
+    )
+
+    if status_filter:
+        stmt = stmt.where(MarketplaceListing.status == status_filter)
+    if seller_id is not None:
+        stmt = stmt.where(MarketplaceListing.seller_id == seller_id)
+    if item_type is not None:
+        stmt = stmt.where(CatalogItem.item_type == item_type)
+    if min_price is not None:
+        stmt = stmt.where(MarketplaceListing.price >= min_price)
+    if max_price is not None:
+        stmt = stmt.where(MarketplaceListing.price <= max_price)
+    if q:
+        search_pattern = f"%{q.strip()}%"
+        stmt = stmt.where(
+            or_(
+                CatalogItem.name.ilike(search_pattern),
+                CatalogItem.description.ilike(search_pattern),
+            )
+        )
+
+    # Count
+    count_stmt = select(func.count()).select_from(stmt.subquery())
+    total = (await db.execute(count_stmt)).scalar() or 0
+
+    stmt = stmt.order_by(MarketplaceListing.created_at.desc(), MarketplaceListing.id.desc()).offset(offset).limit(limit)
+    rows = (await db.execute(stmt)).all()
+
+    listings = [
+        ListingResponse(
+            id=listing.id,
+            catalog_item_id=listing.catalog_item_id,
+            seller_id=listing.seller_id,
+            price=listing.price,
+            status=listing.status,
+            created_at=listing.created_at,
+            updated_at=listing.updated_at,
+            catalog_item=CatalogItemResponse.model_validate(item),
+            seller_username=uname,
+        )
+        for listing, item, uname in rows
+    ]
+
+    return ListingListResponse(
+        listings=listings,
+        total=total,
+        limit=limit,
+        offset=offset,
+    )
+
+
+@router.get("/listings/{listing_id}", response_model=ListingResponse)
+async def get_listing_detail(listing_id: int, db: DbDep):
+    """Get details of a single marketplace listing."""
+    stmt = (
+        select(MarketplaceListing, CatalogItem, User.username)
+        .join(CatalogItem, MarketplaceListing.catalog_item_id == CatalogItem.id)
+        .join(User, MarketplaceListing.seller_id == User.id)
+        .where(MarketplaceListing.id == listing_id)
+    )
+    res = (await db.execute(stmt)).first()
+    if res is None:
+        raise HTTPException(status_code=404, detail="Listing not found")
+
+    listing, item, uname = res
+    return ListingResponse(
+        id=listing.id,
+        catalog_item_id=listing.catalog_item_id,
+        seller_id=listing.seller_id,
+        price=listing.price,
+        status=listing.status,
+        created_at=listing.created_at,
+        updated_at=listing.updated_at,
+        catalog_item=CatalogItemResponse.model_validate(item),
+        seller_username=uname,
+    )
+
+
+@router.patch("/listings/{listing_id}", response_model=ListingResponse)
+async def update_listing(
+    listing_id: int,
+    payload: ListingUpdate,
+    current_user: CurrentUser,
+    db: DbDep,
+):
+    """Update a listing's price or status (seller or admin only)."""
+    stmt = (
+        select(MarketplaceListing, CatalogItem, User.username)
+        .join(CatalogItem, MarketplaceListing.catalog_item_id == CatalogItem.id)
+        .join(User, MarketplaceListing.seller_id == User.id)
+        .where(MarketplaceListing.id == listing_id)
+    )
+    res = (await db.execute(stmt)).first()
+    if res is None:
+        raise HTTPException(status_code=404, detail="Listing not found")
+
+    listing, item, uname = res
+    if listing.seller_id != current_user.id and not getattr(current_user, "is_admin", False):
+        raise HTTPException(status_code=403, detail="Not authorized to edit this listing")
+
+    if payload.price is not None:
+        listing.price = payload.price
+    if payload.status is not None:
+        listing.status = payload.status
+
+    listing.updated_at = datetime.now()
+    await db.commit()
+    await db.refresh(listing)
+
+    return ListingResponse(
+        id=listing.id,
+        catalog_item_id=listing.catalog_item_id,
+        seller_id=listing.seller_id,
+        price=listing.price,
+        status=listing.status,
+        created_at=listing.created_at,
+        updated_at=listing.updated_at,
+        catalog_item=CatalogItemResponse.model_validate(item),
+        seller_username=uname,
+    )
+
+
+@router.delete("/listings/{listing_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_or_delist_listing(
+    listing_id: int,
+    current_user: CurrentUser,
+    db: DbDep,
+):
+    """Delist or delete a listing (seller or admin only)."""
+    listing = await db.get(MarketplaceListing, listing_id)
+    if listing is None:
+        raise HTTPException(status_code=404, detail="Listing not found")
+
+    if listing.seller_id != current_user.id and not getattr(current_user, "is_admin", False):
+        raise HTTPException(status_code=403, detail="Not authorized to delete this listing")
+
+    listing.status = ListingStatus.delisted
+    listing.updated_at = datetime.now()
+    await db.commit()
+    return None
diff --git a/backend/tests/test_marketplace_api.py b/backend/tests/test_marketplace_api.py
new file mode 100644
index 0000000..c594532
--- /dev/null
+++ b/backend/tests/test_marketplace_api.py
@@ -0,0 +1,167 @@
+"""Integration tests for Marketplace Listings API.
+
+docs/planning/economy.md US-3 (task f51c7d95e237).
+"""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+
+def _register(client: TestClient, username: str) -> tuple[int, dict[str, str]]:
+    resp = client.post(
+        "/api/auth/register",
+        json={
+            "username": username,
+            "email": f"{username}@example.com",
+            "password": "secret123",
+            "date_of_birth": "1990-01-01",
+        },
+    )
+    assert resp.status_code == 201, resp.text
+    token = resp.json()["access_token"]
+    headers = {"Authorization": f"Bearer {token}"}
+    me = client.get("/api/users/me", headers=headers)
+    assert me.status_code == 200, me.text
+    return me.json()["id"], headers
+
+
+def test_catalog_items_and_listings_crud():
+    client = TestClient(app)
+    uid1, h1 = _register(client, "mkt_user_1")
+    uid2, h2 = _register(client, "mkt_user_2")
+
+    # 1. Create catalog item by user 1
+    resp = client.post(
+        "/api/marketplace/items",
+        headers=h1,
+        json={
+            "name": "Wizard Hat",
+            "description": "A pointy blue hat with stars",
+            "item_type": "avatar_item",
+            "metadata_json": {"color": "blue", "rarity": "rare"},
+        },
+    )
+    assert resp.status_code == 201, resp.text
+    item_data = resp.json()
+    assert item_data["name"] == "Wizard Hat"
+    assert item_data["owner_id"] == uid1
+    item_id = item_data["id"]
+
+    # 2. List catalog items
+    resp = client.get("/api/marketplace/items", headers=h1, params={"mine": "true"})
+    assert resp.status_code == 200
+    my_items = resp.json()
+    assert len(my_items) >= 1
+    assert any(i["id"] == item_id for i in my_items)
+
+    resp = client.get("/api/marketplace/items", headers=h2, params={"mine": "true"})
+    assert resp.status_code == 200
+    assert not any(i["id"] == item_id for i in resp.json())
+
+    # 3. User 2 tries to list User 1's item -> 403
+    resp = client.post(
+        "/api/marketplace/listings",
+        headers=h2,
+        json={
+            "catalog_item_id": item_id,
+            "price": 100,
+        },
+    )
+    assert resp.status_code == 403, resp.text
+
+    # Invalid price <= 0 -> 422
+    resp = client.post(
+        "/api/marketplace/listings",
+        headers=h1,
+        json={
+            "catalog_item_id": item_id,
+            "price": 0,
+        },
+    )
+    assert resp.status_code == 422
+
+    # 4. User 1 successfully lists item
+    resp = client.post(
+        "/api/marketplace/listings",
+        headers=h1,
+        json={
+            "catalog_item_id": item_id,
+            "price": 250,
+        },
+    )
+    assert resp.status_code == 201, resp.text
+    listing = resp.json()
+    assert listing["price"] == 250
+    assert listing["seller_id"] == uid1
+    assert listing["status"] == "active"
+    assert listing["catalog_item"]["name"] == "Wizard Hat"
+    listing_id = listing["id"]
+
+    # 5. Search / browse listings
+    resp = client.get("/api/marketplace/listings", params={"q": "Wizard"})
+    assert resp.status_code == 200, resp.text
+    data = resp.json()
+    assert data["total"] >= 1
+    assert any(l["id"] == listing_id for l in data["listings"])
+
+    # Filter by item_type
+    resp = client.get("/api/marketplace/listings", params={"item_type": "avatar_item"})
+    assert resp.status_code == 200
+    assert any(l["id"] == listing_id for l in resp.json()["listings"])
+
+    resp = client.get("/api/marketplace/listings", params={"item_type": "experience_pass"})
+    assert resp.status_code == 200
+    assert not any(l["id"] == listing_id for l in resp.json()["listings"])
+
+    # Price range filter
+    resp = client.get("/api/marketplace/listings", params={"min_price": 300})
+    assert resp.status_code == 200
+    assert not any(l["id"] == listing_id for l in resp.json()["listings"])
+
+    resp = client.get("/api/marketplace/listings", params={"max_price": 260})
+    assert resp.status_code == 200
+    assert any(l["id"] == listing_id for l in resp.json()["listings"])
+
+    # 6. Listing detail
+    resp = client.get(f"/api/marketplace/listings/{listing_id}")
+    assert resp.status_code == 200
+    assert resp.json()["id"] == listing_id
+
+    # 7. Update listing
+    # User 2 cannot update User 1's listing
+    resp = client.patch(
+        f"/api/marketplace/listings/{listing_id}",
+        headers=h2,
+        json={"price": 500},
+    )
+    assert resp.status_code == 403
+
+    # User 1 updates price
+    resp = client.patch(
+        f"/api/marketplace/listings/{listing_id}",
+        headers=h1,
+        json={"price": 350},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["price"] == 350
+
+    # 8. Delist listing
+    # User 2 cannot delist
+    resp = client.delete(f"/api/marketplace/listings/{listing_id}", headers=h2)
+    assert resp.status_code == 403
+
+    # User 1 delists
+    resp = client.delete(f"/api/marketplace/listings/{listing_id}", headers=h1)
+    assert resp.status_code == 204
+
+    # Now detail shows delisted
+    resp = client.get(f"/api/marketplace/listings/{listing_id}")
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "delisted"
+
+    # Default search only shows active
+    resp = client.get("/api/marketplace/listings")
+    assert resp.status_code == 200
+    assert not any(l["id"] == listing_id for l in resp.json()["listings"])