Skip to content
Estaite MCP

Lease renewal advisor

A real-world workflow that calls four Estaite tools in one session and synthesizes the results into a recommendation. No LLM required — just rules over the returned numbers. Drop this into a chat agent or a landlord/tenant app as-is.

Scenario: a tenant is up for lease renewal. They tell you their current submarket, current rent, property type, and bedrooms. The script answers: should they accept the renewal, negotiate, or move?

The workflow makes 4 tool calls (search → snapshot → trends → affordability) so we pace them with asyncio.sleep(0.6) between each to stay under the Free tier’s 2 tools/sec limit.

The code

Save as lease-renewal.py:

"""
Lease renewal advisor.


Setup:
    pip install mcp
    export ESTAITE_API_KEY="mk_..."


Usage:
    python lease-renewal.py --submarket "Carmel Valley" \
        --rent 3200 --property-type apt --bedrooms 2
"""


import argparse
import asyncio
import json
import os
import sys


from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client




MCP_URL = "https://mcp.estaite.com"


# Map the user-facing property_type code to the segment label that
# Estaite uses in its response JSON.
SEGMENT_LABEL = {
    "apt": "apartments",
    "sfr": "single_family",
    "ct":  "condo_townhome",
}




def parse_tool_result(result, tool_label):
    """Parse a tool result safely. Returns the JSON payload, or raises with
    a useful message if the tool errored or returned non-JSON."""
    if getattr(result, "isError", False):
        msg = result.content[0].text if result.content else "(no detail)"
        raise RuntimeError(f"{tool_label} returned an error: {msg}")
    if not result.content:
        raise RuntimeError(f"{tool_label} returned an empty response")
    text = result.content[0].text or ""
    if not text.strip():
        raise RuntimeError(f"{tool_label} returned empty text")
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        raise RuntimeError(f"{tool_label} returned non-JSON (first 200 chars): {text[:200]!r}") from e




async def gather(api_key, submarket_name, property_type, bedrooms):
    """Resolve the submarket and pull snapshot + trends + affordability."""
    headers = {"x-api-key": api_key}
    async with streamablehttp_client(MCP_URL, headers=headers) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()


            # Free tier allows 2 tools/sec. We make 4 tool calls in this
            # workflow (search + snapshot + trends + affordability), so we
            # pace ~0.6s between them to stay under the limit. Production
            # code should also catch 429 and honor the Retry-After header.


            # Step 1 — resolve the submarket name to an ID.
            search_result = await session.call_tool(
                "search_estaite_submarkets",
                {"query": submarket_name},
            )
            search = parse_tool_result(search_result, "search_estaite_submarkets")
            results = search.get("results") or []
            if not results:
                raise RuntimeError(f"No submarket matched '{submarket_name}'")
            sm = results[0]
            sm_id, sm_name = sm["id"], sm["name"]
            print(f"  Resolved '{submarket_name}' → {sm_name} (id {sm_id})")
            await asyncio.sleep(0.6)


            # Step 2 — pull three independent views.
            # NB: all three tools take `id` (not `mcp_submarket_id`) in their
            # MCP input schema.
            snapshot_result = await session.call_tool(
                "get_estaite_market_snapshot",
                {"id": sm_id},
            )
            snapshot = parse_tool_result(snapshot_result, "get_estaite_market_snapshot")
            await asyncio.sleep(0.6)


            trends_result = await session.call_tool(
                "get_estaite_rent_trends",
                {
                    "id": sm_id,
                    "property_type": property_type,
                    "bedrooms": bedrooms,
                    "history_months": 12,
                },
            )
            trends = parse_tool_result(trends_result, "get_estaite_rent_trends")
            await asyncio.sleep(0.6)


            afford_result = await session.call_tool(
                "get_estaite_affordability",
                {
                    "id": sm_id,
                    "property_type": property_type,
                    "bedrooms": bedrooms,
                },
            )
            afford = parse_tool_result(afford_result, "get_estaite_affordability")


            return {
                "submarket": {"id": sm_id, "name": sm_name},
                "snapshot":  snapshot,
                "trends":    trends,
                "afford":    afford,
            }




def recommend(current_rent: float, market_median: float, yoy: float) -> tuple[str, str]:
    """Rule-based recommendation. Returns (verdict, reasoning).


    `yoy` is whole-percent (e.g. 3.4 means 3.4%)."""
    if market_median is None:
        return "INSUFFICIENT_DATA", "Couldn't get a market median for this segment."


    delta_pct = (current_rent - market_median) / market_median * 100
    yoy_pct = yoy or 0


    if delta_pct > 10:
        return (
            "NEGOTIATE_OR_MOVE",
            f"You're paying {delta_pct:+.1f}% above market median (${market_median:,.0f}). "
            f"Strong leverage to negotiate down — or move to a comparable unit at market rate.",
        )
    if delta_pct < -10:
        return (
            "ACCEPT_RENEWAL",
            f"You're paying {delta_pct:+.1f}% below market median (${market_median:,.0f}). "
            f"Lock in the renewal — your landlord is unlikely to offer this rate to a new tenant. "
            f"Market is moving {yoy_pct:+.1f}% YoY.",
        )
    return (
        "MARKET_RATE",
        f"You're within ±10% of the market median (${market_median:,.0f}). "
        f"Renewal at flat or slight increase is reasonable. "
        f"Market is moving {yoy_pct:+.1f}% YoY — factor that into negotiations.",
    )




def print_report(args, data):
    sm = data["submarket"]
    segment = SEGMENT_LABEL[args.property_type]
    br_key = f"br{args.bedrooms}"


    # The snapshot's segments use median_rent + rent_change_yoy keys.
    snap_seg = ((data["snapshot"].get("segments") or {}).get(segment) or {}).get(br_key) or {}
    median_rent = snap_seg.get("median_rent")
    yoy         = snap_seg.get("rent_change_yoy")


    afford       = data["afford"]
    afford_label = afford.get("affordability_label") or "—"
    rti_pct      = afford.get("rent_to_income_pct")  # whole-percent


    verdict, reasoning = recommend(args.rent, median_rent, yoy)


    print()
    print(f"  ── Lease renewal advisor: {sm['name']} ──")
    print()
    print(f"  Tenant pays:           ${args.rent:,.0f}/mo ({args.property_type} {args.bedrooms}BR)")
    if median_rent is not None:
        print(f"  Market median:         ${median_rent:,.0f}/mo")
    else:
        print(f"  Market median:         — (no data for {args.property_type} {args.bedrooms}BR)")
    if yoy is not None:
        print(f"  YoY trend:             {yoy:+.1f}%")
    afford_line = f"  Affordability:         {afford_label}"
    if rti_pct is not None:
        afford_line += f" ({rti_pct:.0f}% rent-to-income)"
    print(afford_line)
    print()
    print(f"  Recommendation:        {verdict}")
    print(f"  {reasoning}")
    print()




def main():
    parser = argparse.ArgumentParser(description="Lease renewal advisor for a US tenant")
    parser.add_argument("--submarket",     required=True, help='Submarket name (e.g. "Carmel Valley")')
    parser.add_argument("--rent",          required=True, type=float, help="Current monthly rent (USD)")
    parser.add_argument("--property-type", default="apt", choices=["apt", "sfr", "ct"], help="apt, sfr, or ct")
    parser.add_argument("--bedrooms",      default=2, type=int, choices=[1, 2, 3, 4])
    args = parser.parse_args()


    api_key = os.environ.get("ESTAITE_API_KEY")
    if not api_key:
        print("ERROR: ESTAITE_API_KEY env var not set.")
        sys.exit(1)


    print(f"\nAnalyzing {args.submarket} for {args.property_type} {args.bedrooms}BR...")
    data = asyncio.run(gather(api_key, args.submarket, args.property_type, args.bedrooms))
    print_report(args, data)




if __name__ == "__main__":
    main()

Run it

Terminal window
python lease-renewal.py --submarket "Carmel Valley" --rent 3200 --property-type apt --bedrooms 2

Sample output

Analyzing Carmel Valley for apt 2BR...
  Resolved 'Carmel Valley' → Carmel Valley (id 1)


  ── Lease renewal advisor: Carmel Valley ──


  Tenant pays:           $3,200/mo (apt 2BR)
  Market median:         $3,659/mo
  YoY trend:             -3.2%
  Affordability:         Affordable (23% rent-to-income)


  Recommendation:        ACCEPT_RENEWAL
  You're paying -12.5% below market median ($3,659). Lock in the renewal — your landlord is unlikely to offer this rate to a new tenant. Market is moving -3.2% YoY.

(Numbers shift each month as the data refreshes — yours may differ.)

What’s happening

The script chains four tool calls in one session:

  1. search_estaite_submarkets — resolve the user’s submarket name to an ID. Response contains a results array; we take results[0].
  2. get_estaite_market_snapshot — pulls the median rent and YoY for every property-type / bedroom segment. The segments live at segments[<label>][<br_key>] where label is apartments/single_family/condo_townhome and br_key is br1–br4. Each segment has median_rent (note: _rent here, not _price) and rent_change_yoy (note: NOT plain yoy).
  3. get_estaite_rent_trends — historical trend (used here to confirm direction; you could plot it).
  4. get_estaite_affordability — returns affordability_label (e.g. “Affordable”, “Cost Burdened”) and rent_to_income_pct as whole-percent.

Then it applies a simple rule:

  • >10% above market → negotiate or move
  • >10% below market → lock in the renewal
  • Within ±10% → market rate, factor in trend direction

The rule lives in recommend() — swap it for an LLM call, your own heuristic, or a more nuanced model whenever you want.

A note on the id parameter

The get_estaite_market_snapshot, get_estaite_rent_trends, and get_estaite_affordability tools all accept id (just id) in their MCP input schema, even though the parameter conceptually refers to a submarket id. Don’t pass mcp_submarket_id — that’s the internal function parameter name in the server source, not the wire-protocol key.

Why agent pattern

This is the shape most useful AI applications take: gather a few related data points in parallel or sequence, synthesize, decide. The Estaite tools are designed to compose this way — every *_estaite_* tool returns structured JSON you can feed into the next step or a final language-model summarization.

Next steps

  • Hand the gathered data to an LLM. Replace recommend() with a call to Anthropic, OpenAI, or any other LLM. Pass the snapshot, trends, and affordability as context, ask it to write the renewal email.
  • Run multiple submarkets in parallel. Wrap gather() calls in asyncio.gather() to compare a tenant’s current submarket against three nearby ones (move-suggestion mode). Mind the rate limit — three parallel workflows = 12 tool calls in flight, easily over the Free tier’s 2/sec.
  • Persist past advice. Cache the snapshot per submarket so you only re-fetch when the underlying month changes (Estaite data refreshes monthly).