[truncated — 10919 more characters]
```
### config.py
```python
"""Application configuration loaded from environment variables."""
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""Central configuration for Vetted."""
# Required API keys
anthropic_api_key: str = ""
brightdata_api_key: str = ""
# Optional Bright Data: zone for SERP/Request API; dataset IDs for Web Scraper API
brightdata_zone: str = ""
brightdata_amazon_search_dataset_id: str = "gd_lwdb4vjm1ehb499uxs" # Amazon keyword search
brightdata_amazon_product_dataset_id: str = "gd_l7q7dkf244hwjntr0" # Amazon product page
brightdata_amazon_reviews_dataset_id: str = "gd_le8e811kzy4ggddlq" # Amazon reviews
browserbase_api_key: str = ""
browserbase_project_id: str = ""
perplexity_api_key: str = ""
openai_api_key: str = "" # OpenAI GPT-4o for negotiation agent
# Optional
elastic_cloud_id: str = ""
elastic_api_key: str = ""
# App config
log_level: str = "INFO"
agent_timeout: int = 120
enable_negotiation: bool = True
model_config = {
"env_file": (".env", "../.env"), # check shopagent/.env first, then repo root
"env_file_encoding": "utf-8",
}
settings = Settings()
```
### test_search_agent.py
```python
"""Quick end-to-end test for the SearchAgent."""
import asyncio
import logging
import sys
# Load .env before anything else imports settings
from dotenv import load_dotenv
load_dotenv()
from config import settings
from models.requirements import ProductRequirements
from models.state import SharedState
from agents.search_agent import SearchAgent
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
async def main() -> None:
# 1. Check API key status
has_bright = bool(settings.brightdata_api_key and settings.brightdata_api_key.strip())
print(f"\n=== Search Agent E2E Test ===")
print(f"Bright Data API key configured: {has_bright}")
print(f"Bright Data zone: {settings.brightdata_zone or '(none)'}")
print(f"Amazon search dataset ID: {settings.brightdata_amazon_search_dataset_id or '(none)'}")
print(f"Amazon product dataset ID: {settings.brightdata_amazon_product_dataset_id or '(none)'}")
print(f"Amazon reviews dataset ID: {settings.brightdata_amazon_reviews_dataset_id or '(none)'}")
# 2. Build requirements
reqs = ProductRequirements(
category="mirrorless camera",
description="mirrorless camera for travel photography",
must_have=["mirrorless", "4K video"],
nice_to_have=["image stabilization", "weather sealed", "compact"],
dealbreakers=["DSLR only"],
budget_min=None,
budget_max=800.0,
brand_preferences=["Sony", "Fujifilm"],
brand_exclusions=[],
use_case="travel photography",
urgency="this_month",
condition="new",
)
# 3. Minimal shared state
state = SharedState(
session_id="test-001",
user_query="mirrorless camera, budget $500-800, Sony or Fujifilm, for travel photography",
requirements=reqs,
requirements_finalized=True,
status="searching",
)
# 4. Run the search agent
from agents.search_agent import _build_search_query
print(f"\nBuilt search query: \"{_build_search_query(reqs)}\"")
agent = SearchAgent()
print("Running SearchAgent...")
result = await agent.run(state)
# 5. Print results
candidates = result.get("candidates", [])
print(f"\n--- Results ---")
print(f"Status: {result.get('status')}")
print(f"Candidates returned: {len(candidates)}")
if result.get("errors"):
print(f"Errors: {result['errors']}")
for i, c in enumerate(candidates, 1):
print(f"\n {i}. {c.name}")
print(f" Brand: {c.brand} | Price: ${c.price:.2f} | Platform: {c.platform}")
print(f" Rating: {c.rating} | Reviews: {c.review_count}")
print(f" URL: {c.url}")
if c.matched_requirements:
print(f" Matched reqs: {c.matched_requirements}")
# 6. Detect mock vs real
mock_indicators = sum(1 for c in candidates if "Mock" in c.name)
if mock_indicators == len(candidates) and candidates:
print(f"\n>>> Data source: MOCK (all {len(candidates)} candidates are mock data)")
elif mock_indicators > 0:
print(f"\n>>> Data source: MIXED ({len(candidates) - mock_indicators} real, {mock_indicators} mock fallback)")
else:
print(f"\n>>> Data source: REAL API (no mock candidates detected)")
if __name__ == "__main__":
asyncio.run(main())
```
### test_trust_agent.py
```python
"""Quick end-to-end test for the TrustAgent with real Perplexity Sonar API."""
import asyncio
import logging
from dotenv import load_dotenv
load_dotenv()
from config import settings
from models.candidates import ProductCandidate
from models.state import SharedState
from agents.trust_agent import TrustAgent
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
async def main() -> None:
has_pplx = bool(settings.perplexity_api_key and settings.perplexity_api_key.strip())
print(f"\n=== Trust Agent E2E Test ===")
print(f"Perplexity API key configured: {has_pplx}")
# Build a small set of realistic candidates (2 real-ish, 1 suspicious)
candidates = [
ProductCandidate(
id="c001",
name="Sony Alpha ZV-E10 Mirrorless Camera",
brand="Sony",
price=798.00,
url="https://www.amazon.com/dp/B0FLSPG85G",
platform="amazon",
seller_name="Amazon.com",
rating=4.6,
review_count=905,
),
ProductCandidate(
id="c002",
name="Canon EOS R100 Mirrorless Camera Kit",
brand="Canon",
price=715.89,
url="https://www.amazon.com/dp/B0FP496DLW",
platform="amazon",
seller_name="Camera Photo Photo",
rating=4.5,
review_count=35,
),
ProductCandidate(
id="c003",
name="ProShot 8K Ultra Camera AMAZING DEAL",
brand="Unknown",
price=29.99,
url="https://www.amazon.com/dp/B0FAKE12345",
platform="amazon",
seller_name="BestDealzXtreme",
rating=5.0,
review_count=12000,
),
]
state = SharedState(
session_id="trust-test-001",
user_query="mirrorless camera for travel",
candidates=candidates,
status="analyzing",
)
agent = TrustAgent()
print(f"\nRunning TrustAgent on {len(candidates)} candidates...\n")
result = await agent.run(state)
trust_scores = result.get("trust_scores", {})
print(f"--- Results: {len(trust_scores)} trust scores ---\n")
for cid, score in trust_scores.items():
c = next((x for x in candidates if x.id == cid), None)
name = c.name[:50] if c else cid
print(f" [{cid}] {name}")
print(f" Overall: {score.overall_score}/100")
print(f" Seller: {score.seller_score}/100")
print(f" Reviews: {score.review_authenticity_score}/100")
print(f" Legit: {score.product_legitimacy_score}/100")
if score.flags:
for f in score.flags:
print(f" FLAG [{f.severity}] {f.category}: {f.description}")
print(f" Reasoning: {score.reasoning[:200]}...")
print(f" Sources: {score.sources_checked[:3]}")
print()
# Check if we got real or mock data
mock_count = sum(
1 for s in trust_scores.values() if "Mock" in s.reasoning
)
if mock_count == len(trust_scores):
print(">>> Data source: MOCK (all scores are mock)")
elif mock_count > 0:
print(f">>> Data source: MIXED ({len(trust_scores) - mock_count} real, {mock_count} mock)")
else:
print(">>> Data source: REAL API (all scores from Perplexity Sonar)")
if __name__ == "__main__":
asyncio.run(main())
```
[67 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]