Pierce the Corporate Veil with AI

Resolve the real humans behind NYC real estate LLCs using public data APIs.

Introduction

30 Second Summary

Every day, billions of dollars in NYC real estate changes hands through LLCs. The real buyers and sellers hide behind corporate names, making it nearly impossible to know who actually owns what.

In this project, you will build a Python tool that "pierces the corporate veil" by cross-referencing NYC property transaction records with state filings to reveal the actual humans behind LLC deals.

What You'll Build

A CLI tool and serverless API that resolves LLC ownership from NYC property records using public datasets and AI-powered document extraction.

By the end of this project, you'll have:

  • A working ACRIS transaction search and LLC ownership resolver that cross-references NYC property records with NY Secretary of State filings via the Socrata Open Data API.
  • A Claude Vision module that extracts structured buyer/seller data from scanned deed images using forced tool_use for guaranteed JSON output.
  • A deployed AWS Lambda endpoint exposing the full pipeline as a serverless API that costs $0 when idle.
  • Secret Mission: Add comparable sales calculation that estimates property value based on recent ACRIS transactions in the same block.

Are there any prerequisites?

You'll need Python 3.10+ installed, an AWS account with the CLI configured, and an Anthropic API key. No prior experience with NYC Open Data or the Socrata API is required.

Before We Start

Most NYC real estate buyers hide behind LLCs. A deed might say "123 MAIN STREET LLC" bought a $5 million building, but the actual human making decisions stays invisible.

Your tool will pierce that corporate veil by querying two public datasets. Before you can build it, you need a working Python environment with the right dependencies and verified access to the APIs.

In this step, get ready to:
  • Set up a Python virtual environment and install dependencies.
  • Configure your Anthropic API key and verify the SDK connects.
  • Verify access to NYC Open Data and NY State Open Data APIs.
Create your project directory and install dependencies

Your project needs two Python packages. The anthropic SDK communicates with Claude for document extraction. The requests library handles HTTP calls to the public data APIs.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Terminal, and pressing Enter.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir llc-resolver
cd llc-resolver
  • Create a Python virtual environment by running:
python3 -m venv venv

What is a virtual environment?

A virtual environment is an isolated Python workspace. It keeps your project's packages separate from your system Python, so different projects can use different package versions without conflicts.

  • Activate the virtual environment:

macOS

source venv/bin/activate

Windows

.\venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt.

Don't see (venv) in your prompt?

Make sure you ran the activation command from inside the llc-resolver folder. If you see a "No such file or directory" error, run ls to confirm the venv folder exists.

Help me activate my Python virtual environment.

  • Create the requirements.txt file that lists your project dependencies by running:
cat << 'EOF' > requirements.txt
anthropic==0.104.1
requests==2.34.2
EOF
  • Install the dependencies by running:
pip install -r requirements.txt

You should see output showing both packages downloading and installing successfully.

Seeing installation errors?

Make sure your virtual environment is active (you should see (venv) in your prompt). If pip is not found, try pip3 instead.

Help me install Python packages with pip.

  • Confirm the files exist by running:
ls

You should see requirements.txt and the venv folder in the output.

✔️ Awesome, I've got everything!

Great. Make sure you've saved requirements.txt and your virtual environment is active before continuing.

ⓧ I'd like to double check the full code

anthropic==0.104.1
requests==2.34.2
Configure your Anthropic API key

The Anthropic SDK needs your API key to authenticate requests to Claude. You'll set it as an environment variable so the SDK picks it up automatically.

  • Set your ANTHROPIC_API_KEY environment variable by running the following command, replacing your-api-key-here with your actual Anthropic API key:

macOS

export ANTHROPIC_API_KEY=your-api-key-here

Windows

$env:ANTHROPIC_API_KEY="your-api-key-here"

Keep your key safe

This environment variable only lasts for your current terminal session. You'll need to set it again if you close and reopen your terminal.

  • Verify that the Anthropic SDK connects successfully by running this test command:
python3 -c "import anthropic; client = anthropic.Anthropic(); print('Anthropic SDK connected successfully')"

You should see Anthropic SDK connected successfully printed in your terminal.

Seeing an authentication error?

Double-check that you replaced your-api-key-here with your actual API key (starting with sk-ant-). Make sure there are no extra spaces around the key value.

Help me connect the Anthropic SDK.

Verify access to the public data APIs

Your resolver queries two public datasets. The ACRIS (Automated City Register Information System) dataset on NYC Open Data holds all property transaction records. The NY Secretary of State dataset on data.ny.gov holds LLC filings with owner details.

Both use the Socrata Open Data API (SODA), which is free and requires no API key.

  • Test the ACRIS Real Property Parties endpoint by running:
python3 -c "
import requests
response = requests.get('https://data.cityofnewyork.us/resource/636b-3b5g.json', params={'\$limit': '1'})
print(f'ACRIS status: {response.status_code}')
print(f'Sample record: {response.json()[0]["name"] if response.json() else "empty"}')
"

What does this test do?

This queries the ACRIS Real Property Parties dataset (636b-3b5g) with a limit of 1 record. A 200 status code and a returned name confirm you can reach the dataset.

You should see ACRIS status: 200 followed by a party name from the records.

  • Test the NY Secretary of State Active Corporations endpoint by running:
python3 -c "
import requests
response = requests.get('https://data.ny.gov/resource/n9v6-gdp6.json', params={'\$limit': '1'})
print(f'NY DOS status: {response.status_code}')
print(f'Sample entity: {response.json()[0]["current_entity_name"] if response.json() else "empty"}')
"

You should see NY DOS status: 200 followed by a corporation name.

Getting a connection error?

These are public APIs with no authentication required. If you see a timeout or connection refused error, check your internet connection. If you see a 403 or 429 status code, wait a minute and try again.

Help me connect to the Socrata API.

Your environment is ready and all three external services are reachable. Next up, you'll build the ACRIS transaction search module that finds every deal an LLC has been involved in.

Query ACRIS for Property Transactions

In the previous step, you set up your project environment and confirmed access to NYC's public data APIs. Now it's time to build the core search engine that finds every property transaction an LLC has been involved in.

NYC's ACRIS (Automated City Register Information System) stores every recorded property document in the city. It splits data across linked tables: one table for the parties involved, one for the document details. Your job is to query both and stitch the results together into a transaction history.

In this step, get ready to:
  • Build a party search function that queries ACRIS for all records matching a name.
  • Build a document lookup function that fetches transaction details (dates, amounts, types).
  • Test the pipeline end-to-end with a real LLC name.
Build the ACRIS party search function

The Socrata Open Data API (SODA) powers both NYC Open Data and NY State Open Data. It accepts SoQL queries via GET parameters, which means you can filter millions of records server-side before any data reaches your machine.

  • Create a new file called resolver.py in your project folder. Start with the imports and API endpoint constants by pasting this code:
import anthropic
import requests
import base64
import json
from typing import Optional


# ACRIS Real Property Parties - records every buyer/seller on NYC property docs
ACRIS_PARTIES_ENDPOINT = "https://data.cityofnewyork.us/resource/636b-3b5g.json"
# ACRIS Real Property Master - document details (date, amount, type)
ACRIS_MASTER_ENDPOINT = "https://data.cityofnewyork.us/resource/bnx9-e6tj.json"
# NY Secretary of State Active Corporations - LLC registration data
NY_DOS_ENDPOINT = "https://data.ny.gov/resource/n9v6-gdp6.json"

What does this code do?

The imports bring in everything the full resolver will eventually need. For now, you'll use requests to call the Socrata API.

Each endpoint constant follows the Socrata pattern: https://data.{domain}/resource/{dataset-id}.json. The dataset IDs (636b-3b5g, bnx9-e6tj, n9v6-gdp6) are unique identifiers published on NYC Open Data and NY State Open Data.

  • Below the constants, add the search_acris_parties() function that searches the ACRIS Real Property Parties dataset by name:
def search_acris_parties(name: str, limit: int = 50) -> list:
    """Search ACRIS Real Property Parties by name."""
    # Convert to uppercase and escape single quotes for SoQL safety
    safe_name = name.upper().replace("'", "''")
    params = {
        # Case-insensitive substring match using SoQL upper() and LIKE
        "$where": f"upper(name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
        # Most recent records first
        "$order": "good_through_date DESC",
    }
    response = requests.get(ACRIS_PARTIES_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()

What does this code do?

  • safe_name uppercases the input and escapes single quotes. SoQL on Socrata 2.1 endpoints is case-sensitive, so the upper() function in the $where clause ensures a case-insensitive match.
  • The LIKE '%...%' pattern does a substring search. This catches partial matches like searching "BROADWAY" and finding "123 BROADWAY LLC".
  • $order sorts by good_through_date DESC so the most recent transactions appear first.
  • Save resolver.py.

Seeing a syntax error?

  • Make sure the f-string in $where uses matching single and double quotes. The outer string uses double quotes, and the SoQL LIKE pattern uses single quotes inside.
  • Check that the function is at the top level (no extra indentation). It should start at column 0.

Why is my search_acris_parties function throwing a syntax error?

Build the document lookup function

The parties dataset tells you WHO was on a transaction, but not the details. Each party record has a document_id that links it to the master record containing the document date, dollar amount, and type.

  • Below search_acris_parties(), add the get_acris_document() function that fetches the master record for a given document ID:
def get_acris_document(document_id: str) -> dict:
    """Get the master record for a given ACRIS document ID."""
    # Filter master records by the 16-character document ID
    params = {"$where": f"document_id='{document_id}'"}
    response = requests.get(ACRIS_MASTER_ENDPOINT, params=params)
    response.raise_for_status()
    results = response.json()
    # Return the first match or empty dict if not found
    return results[0] if results else {}

What does this code do?

  • The function takes a 16-character ACRIS document ID and queries the Real Property Master dataset to get the full record.
  • Each document ID maps to one master record containing doc_date, doc_amount, and doc_type (e.g., DEED, MORTGAGE, ASSIGNMENT).
  • The fallback to an empty dict prevents your code from crashing if a document ID has no matching master record.
  • Save resolver.py.

Why two separate functions instead of one big query?

ACRIS stores parties and document details in separate datasets linked by document_id. The Socrata API does not support JOINs across datasets, so you must query them individually and link the results in your Python code.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on to the test.

ⓧ I'd like to double check the full code

import anthropic
import requests
import base64
import json
from typing import Optional


ACRIS_PARTIES_ENDPOINT = "https://data.cityofnewyork.us/resource/636b-3b5g.json"
ACRIS_MASTER_ENDPOINT = "https://data.cityofnewyork.us/resource/bnx9-e6tj.json"
NY_DOS_ENDPOINT = "https://data.ny.gov/resource/n9v6-gdp6.json"


def search_acris_parties(name: str, limit: int = 50) -> list:
    """Search ACRIS Real Property Parties by name."""
    safe_name = name.upper().replace("'", "''")
    params = {
        "$where": f"upper(name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
        "$order": "good_through_date DESC",
    }
    response = requests.get(ACRIS_PARTIES_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()


def get_acris_document(document_id: str) -> dict:
    """Get the master record for a given ACRIS document ID."""
    params = {"$where": f"document_id='{document_id}'"}
    response = requests.get(ACRIS_MASTER_ENDPOINT, params=params)
    response.raise_for_status()
    results = response.json()
    return results[0] if results else {}
Test with a real LLC name

Both functions are in place. Time to verify they work against live ACRIS data by searching for a common NYC property name and then enriching the results with document details.

  • Run this test command in your terminal to search for party records and fetch the first document's details:
python -c "
from resolver import search_acris_parties, get_acris_document
import json

parties = search_acris_parties('BROADWAY')
print(f'Found {len(parties)} party records')

if parties:
    doc_id = parties[0].get('document_id')
    print(f'First document ID: {doc_id}')
    doc = get_acris_document(doc_id)
    print(json.dumps(doc, indent=2, default=str))
"
  • You should see output showing the number of records found, a 16-character document ID, and a JSON object with fields like doc_date, doc_amount, and doc_type.

What should I see?

A successful run produces something like:

Found 50 party records
First document ID: 2024012300345001
{
  "document_id": "2024012300345001",
  "doc_type": "DEED",
  "doc_date": "2024-01-15T00:00:00.000",
  "doc_amount": "1250000",
  ...
}

The exact values will differ because this is live data. What matters is that you see a numeric count, a document ID, and a JSON object with transaction details.

Getting a connection error or empty results?

  • If you see a connection error, check your internet connection. The Socrata API does not require authentication for basic queries, but it does need outbound HTTPS access.
  • If the result is 0 records, try a different search term. "BROADWAY" is a common name in NYC real estate. If your network restricts access to government data sites, try from a different network.
  • If you see ModuleNotFoundError: No module named 'resolver', make sure your terminal is in the same directory as resolver.py.

Why is my ACRIS query returning an error or no results?

Your ACRIS search engine is working. You can now find every property transaction for any name in NYC's records and pull the full details for each deal. Next up, you'll use the NY Secretary of State data to reveal the actual humans behind those LLC names.

Resolve LLC Ownership via NY Secretary of State

Your ACRIS search can now find every transaction an LLC has been involved in. But transaction records only show entity names. They do not reveal the actual humans making decisions behind those entities.

The NY Department of State maintains a public database of every corporation and LLC registered in the state. This dataset includes the CEO/Chairman, registered agent, and service of process contact for each entity. In this step, you will query that dataset to connect LLC names to real people.

In this step, get ready to:
  • Build a function that queries the NY Secretary of State Open Data API.
  • Understand which fields reveal the people behind an LLC.
  • Test the function with a real entity name and view the ownership profile.
Build the NY DOS search function

The NY Secretary of State publishes an Active Corporations dataset on data.ny.gov. This dataset uses the same Socrata API you used for ACRIS, so the query pattern is nearly identical.

  • In resolver.py, scroll to the bottom of the file (after the get_acris_document() function).
  • Add the following function that searches the NY DOS dataset by entity name:
def search_ny_dos(entity_name: str, limit: int = 10) -> list:
    """Search NY DOS Active Corporations by entity name."""
    safe_name = entity_name.upper().replace("'", "''")
    params = {
        "$where": f"upper(current_entity_name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
    }
    response = requests.get(NY_DOS_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()

What does this code do?

This function follows the same pattern as your ACRIS search, but targets a different dataset on a different Socrata host (data.ny.gov instead of data.cityofnewyork.us).

  • safe_name uppercases the input and escapes single quotes for safe SoQL queries.
  • The $where clause uses a LIKE filter on current_entity_name to find partial matches.
  • The $limit is set to 10 since you typically only need the top matches for a given LLC.
  • Save resolver.py. You will test this function in the next substep.

Seeing a syntax error on save?

Make sure the function is not indented inside get_acris_document(). It should be a top-level function at the same indentation as the other functions in the file.

help me fix a syntax error in my search_ny_dos function

Identify the people behind the LLC

The NY DOS dataset returns rich entity records. Each record contains fields that reveal the real humans associated with the LLC.

What ownership fields does the API return?

Each entity record contains up to three contact roles that help you identify the people behind the corporate veil:

  • chairman_name and chairman_address_1 reveal the CEO or Chairman. This is often the actual decision-maker.
  • dos_process_name and dos_process_address_1 show the Service of Process contact. This is the person designated to receive legal documents.
  • registered_agent_name and registered_agent_address_1 identify the Registered Agent. This is sometimes a law firm or agent service, but can also be the principal themselves.

Why three separate roles?

New York requires LLCs to designate a Service of Process contact and optionally a Registered Agent. The CEO/Chairman is the filed managing member. In many small NYC property LLCs, all three are the same person. In larger operations, the registered agent might be a service like CT Corporation while the CEO field reveals the actual owner.

Test the function with a real LLC name
  • Run this command in your terminal to query the NY Secretary of State for entities with "BROADWAY" in their name:
python3 << 'EOF'
from resolver import search_ny_dos
import json

results = search_ny_dos("BROADWAY")
if results:
    e = results[0]
    print(f"Entity: {e.get('current_entity_name')}")
    print(f"Type: {e.get('entity_type')}")
    print(f"Filing Date: {e.get('initial_dos_filing_date')}")
    print(f"CEO/Chairman: {e.get('chairman_name', 'N/A')}")
    print(f"Service of Process: {e.get('dos_process_name', 'N/A')}")
    print(f"Registered Agent: {e.get('registered_agent_name', 'N/A')}")
    print(f"\nFull first record:")
    print(json.dumps(e, indent=2, default=str))
else:
    print("No results found")
EOF

What should I see?

You should see an entity name containing "BROADWAY", along with at least one of the three people fields (CEO/Chairman, Service of Process, or Registered Agent) populated with a name. The full JSON output shows all available fields for the first matching record.

If a field shows N/A, that role was not filed for this particular entity. Try a different search term if all three are empty.

  • Try replacing "BROADWAY" with an LLC name you found in your ACRIS transaction results from the previous step.

Getting an empty result or connection error?

  • If you see "No results found", try a broader search term. Single common words like "BROADWAY" or "PARK" reliably return results from the NY DOS dataset.
  • If you see a connection error, check that your virtual environment is active and requests is installed.
  • If you get a throttling error, the Socrata API limits unauthenticated requests. Wait a few seconds and try again.

help me debug why my NY DOS search is not returning results

✔️ Awesome, I've got everything!

Great. Double check you have saved resolver.py with the new search_ny_dos() function.

ⓧ I'd like to double check the full code

import anthropic
import requests
import base64
import json
from typing import Optional


ACRIS_PARTIES_ENDPOINT = "https://data.cityofnewyork.us/resource/636b-3b5g.json"
ACRIS_MASTER_ENDPOINT = "https://data.cityofnewyork.us/resource/bnx9-e6tj.json"
NY_DOS_ENDPOINT = "https://data.ny.gov/resource/n9v6-gdp6.json"


def search_acris_parties(name: str, limit: int = 50) -> list:
    """Search ACRIS Real Property Parties by name."""
    safe_name = name.upper().replace("'", "''")
    params = {
        "$where": f"upper(name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
        "$order": "good_through_date DESC",
    }
    response = requests.get(ACRIS_PARTIES_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()


def get_acris_document(document_id: str) -> dict:
    """Get the master record for a given ACRIS document ID."""
    params = {"$where": f"document_id='{document_id}'"}
    response = requests.get(ACRIS_MASTER_ENDPOINT, params=params)
    response.raise_for_status()
    results = response.json()
    return results[0] if results else {}


def search_ny_dos(entity_name: str, limit: int = 10) -> list:
    """Search NY DOS Active Corporations by entity name."""
    safe_name = entity_name.upper().replace("'", "''")
    params = {
        "$where": f"upper(current_entity_name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
    }
    response = requests.get(NY_DOS_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()

You can now find the humans behind any LLC registered in New York. Next, you will build a Claude Vision module that extracts structured buyer and seller data directly from scanned deed images.

Extract Deed Data from Scanned Images with Claude Vision

You can now query two public datasets to find transactions and resolve LLC ownership. But what about the starting point? In real estate due diligence, you often start with a scanned deed image, not a clean LLC name.

In this step, you will build a function that takes a scanned deed image and uses Claude Vision to extract structured data from it. The tool_use pattern guarantees the output matches a strict JSON schema, so you can feed the extracted LLC name directly into your resolver pipeline.

In this step, get ready to:
  • Build an image-loading function that base64-encodes deed scans for the Claude API.
  • Define a JSON schema and force Claude to return structured deed data via tool_use.
  • Test the extraction with a sample deed image and verify the output.
Build the image loader and encoder

Claude Vision accepts images as base64-encoded strings alongside a media type. Your function needs to read the image file from disk, encode it, and detect the correct format.

  • In resolver.py, scroll to the bottom of the file (below the search_ny_dos function).
  • Add the start of the extract_deed_from_image function by pasting this code:
def extract_deed_from_image(image_path: str) -> dict:
    """Extract structured data from a scanned deed image using Claude Vision."""
    client = anthropic.Anthropic()

    # Read the image file and base64-encode it for the API
    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    # Detect the correct media type from the file extension
    if image_path.lower().endswith(".png"):
        media_type = "image/png"
    elif image_path.lower().endswith(".webp"):
        media_type = "image/webp"
    else:
        media_type = "image/jpeg"

What does this code do?

This is the first half of your Claude Vision function. It handles preparing the image for the API.

  • anthropic.Anthropic() creates a client that reads your ANTHROPIC_API_KEY from the environment automatically.
  • base64.standard_b64encode() converts the raw image bytes into a text string that can travel inside a JSON request.
  • The media type detection tells Claude whether to decode the image as PNG, WebP, or JPEG.

You will not see any output yet. The function is incomplete until you add the schema and API call in the next substep.

Define the deed schema and call Claude with tool_use

Claude's tool_use feature lets you define a JSON schema that the model must conform to. By setting tool_choice to force a specific tool, you guarantee schema-conformant output without regex parsing or retries.

  • Directly below the media type detection code you just added, paste the deed schema definition:
    deed_schema = {
        "type": "object",
        "properties": {
            "buyer_name": {
                "type": ["string", "null"],
                "description": "The grantee/buyer name as written on the deed",
            },
            "seller_name": {
                "type": ["string", "null"],
                "description": "The grantor/seller name as written on the deed",
            },
            "property_address": {
                "type": ["string", "null"],
                "description": "The property street address",
            },
            "borough": {
                "type": ["string", "null"],
                "description": "NYC borough (Manhattan, Brooklyn, Queens, Bronx, Staten Island)",
            },
            "block": {
                "type": ["string", "null"],
                "description": "Tax block number",
            },
            "lot": {
                "type": ["string", "null"],
                "description": "Tax lot number",
            },
            "sale_price": {
                "type": ["number", "null"],
                "description": "Sale price in dollars",
            },
            "document_date": {
                "type": ["string", "null"],
                "description": "Date of the deed (YYYY-MM-DD if possible)",
            },
            "is_llc_buyer": {
                "type": "boolean",
                "description": "Whether the buyer name contains LLC, L.L.C., or similar",
            },
        },
        "required": [
            "buyer_name",
            "seller_name",
            "property_address",
            "is_llc_buyer",
        ],
    }

What does this schema do?

The schema defines the exact fields Claude must extract from the deed image. Each property has a type and description that guides the model.

Fields use ["string", "null"] types because not every deed contains all information. The required array ensures Claude always attempts to extract the buyer, seller, address, and LLC status.

  • Now paste the API call that sends the image to Claude with the schema enforced via tool_choice:
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=[
            {
                "name": "extract_deed",
                "description": "Extract structured information from a NYC property deed document.",
                "input_schema": deed_schema,
            }
        ],
        tool_choice={"type": "tool", "name": "extract_deed"},
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media_type,
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": "Extract all available information from this NYC property deed document. Look for the buyer (grantee), seller (grantor), property address, borough/block/lot, sale price, and date.",
                    },
                ],
            }
        ],
    )

    for block in response.content:
        if block.type == "tool_use":
            return block.input
    return {}

What does this code do?

This is where the magic happens. The tool_choice parameter set to {"type": "tool", "name": "extract_deed"} forces Claude to always produce a tool_use response matching your schema.

The message contains both the base64-encoded image and a text prompt guiding extraction. Claude reads the deed image and returns structured data in the block.input field of the tool_use response block.

Why tool_use instead of asking Claude for JSON directly?

When you ask an LLM to return JSON in plain text, it sometimes wraps it in markdown, adds commentary, or produces invalid syntax. The tool_use pattern sidesteps this entirely. Claude is fine-tuned to produce arguments that match the tool's schema exactly.

  • Save resolver.py.

Seeing a syntax error?

Make sure the deed_schema variable, the response = client.messages.create(...) call, and the for block loop are all indented inside the extract_deed_from_image function body (one level of indentation).

help me fix an indentation error in my extract_deed_from_image function

Test with a sample deed image

To verify the extraction works, you need a deed image. You can use a screenshot of any NYC property deed document. ACRIS lets you view document images for properties recorded since 1966.

  • Get a sample deed image using one of these options:
  • Save the image as sample_deed.jpg (or .png) in your llc-resolver project folder.
  • Test the function by running this command:
python -c "from resolver import extract_deed_from_image; import json; result = extract_deed_from_image('sample_deed.jpg'); print(json.dumps(result, indent=2))"
  • You should see structured JSON output with fields like buyer_name, seller_name, property_address, and is_llc_buyer populated from the image.

What should I see?

The output is a JSON object with the fields defined in your schema. Fields that Claude could not find in the image will be null. The is_llc_buyer field will be true or false depending on whether the buyer name contains LLC or similar.

Getting an authentication error?

Make sure your ANTHROPIC_API_KEY environment variable is set in the terminal session where you are running the command. You can verify by running echo $ANTHROPIC_API_KEY (macOS/Linux) or echo %ANTHROPIC_API_KEY% (Windows).

help me debug an authentication error when calling the Claude API from my resolver

✔️ Awesome, I've got everything!

Great. Double check you have saved resolver.py.

ⓧ I'd like to double check the full code

import anthropic
import requests
import base64
import json
from typing import Optional


ACRIS_PARTIES_ENDPOINT = "https://data.cityofnewyork.us/resource/636b-3b5g.json"
ACRIS_MASTER_ENDPOINT = "https://data.cityofnewyork.us/resource/bnx9-e6tj.json"
NY_DOS_ENDPOINT = "https://data.ny.gov/resource/n9v6-gdp6.json"


def search_acris_parties(name: str, limit: int = 50) -> list:
    """Search ACRIS Real Property Parties by name."""
    safe_name = name.upper().replace("'", "''")
    params = {
        "$where": f"upper(name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
        "$order": "good_through_date DESC",
    }
    response = requests.get(ACRIS_PARTIES_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()


def get_acris_document(document_id: str) -> dict:
    """Get the master record for a given ACRIS document ID."""
    params = {"$where": f"document_id='{document_id}'"}
    response = requests.get(ACRIS_MASTER_ENDPOINT, params=params)
    response.raise_for_status()
    results = response.json()
    return results[0] if results else {}


def search_ny_dos(entity_name: str, limit: int = 10) -> list:
    """Search NY DOS Active Corporations by entity name."""
    safe_name = entity_name.upper().replace("'", "''")
    params = {
        "$where": f"upper(current_entity_name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
    }
    response = requests.get(NY_DOS_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()


def extract_deed_from_image(image_path: str) -> dict:
    """Extract structured data from a scanned deed image using Claude Vision."""
    client = anthropic.Anthropic()

    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    if image_path.lower().endswith(".png"):
        media_type = "image/png"
    elif image_path.lower().endswith(".webp"):
        media_type = "image/webp"
    else:
        media_type = "image/jpeg"

    deed_schema = {
        "type": "object",
        "properties": {
            "buyer_name": {
                "type": ["string", "null"],
                "description": "The grantee/buyer name as written on the deed",
            },
            "seller_name": {
                "type": ["string", "null"],
                "description": "The grantor/seller name as written on the deed",
            },
            "property_address": {
                "type": ["string", "null"],
                "description": "The property street address",
            },
            "borough": {
                "type": ["string", "null"],
                "description": "NYC borough (Manhattan, Brooklyn, Queens, Bronx, Staten Island)",
            },
            "block": {
                "type": ["string", "null"],
                "description": "Tax block number",
            },
            "lot": {
                "type": ["string", "null"],
                "description": "Tax lot number",
            },
            "sale_price": {
                "type": ["number", "null"],
                "description": "Sale price in dollars",
            },
            "document_date": {
                "type": ["string", "null"],
                "description": "Date of the deed (YYYY-MM-DD if possible)",
            },
            "is_llc_buyer": {
                "type": "boolean",
                "description": "Whether the buyer name contains LLC, L.L.C., or similar",
            },
        },
        "required": [
            "buyer_name",
            "seller_name",
            "property_address",
            "is_llc_buyer",
        ],
    }

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=[
            {
                "name": "extract_deed",
                "description": "Extract structured information from a NYC property deed document.",
                "input_schema": deed_schema,
            }
        ],
        tool_choice={"type": "tool", "name": "extract_deed"},
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media_type,
                            "data": image_data,
                        },
                    },
                    {
                        "type": "text",
                        "text": "Extract all available information from this NYC property deed document. Look for the buyer (grantee), seller (grantor), property address, borough/block/lot, sale price, and date.",
                    },
                ],
            }
        ],
    )

    for block in response.content:
        if block.type == "tool_use":
            return block.input
    return {}

Your resolver now has eyes. It can read scanned deed images and pull out structured buyer, seller, and property data. Next up, you will wire all three modules together into a single pipeline that takes any LLC name or deed image and produces a full ownership report.

Build the Full Resolver Pipeline

You now have three separate modules: an ACRIS transaction search, a NY Secretary of State lookup, and a Claude Vision deed extractor. Each one works individually, but the real power comes from connecting them into a single pipeline.

In this step, you will build the orchestration function that ties everything together. Given an LLC name (or a scanned deed image), your resolver will automatically extract the entity name, pull its transaction history, and reveal the humans behind it.

In this step, get ready to:
  • Create the main resolve_ownership() function that orchestrates all three modules.
  • Add a CLI interface using argparse for running the resolver from your terminal.
  • Run the full pipeline end-to-end and verify a complete ownership profile.
Create the main resolver function

The resolve_ownership() function is the brain of your tool. It decides the order of operations: if a deed image is provided, extract the LLC name first. Then query ACRIS for transactions. Then look up the entity in the NY Secretary of State records.

  • In resolver.py, scroll to the bottom of the file (after the extract_deed_from_image function) and add the following function:
def resolve_ownership(name: str, image_path: Optional[str] = None) -> dict:
    """Full ownership resolution pipeline."""
    result = {
        "query": name,
        "deed_extraction": None,
        "acris_transactions": [],
        "dos_entity": None,
        "resolved_owners": [],
    }

    # Step 1: If a deed image is provided, extract the buyer name from it
    if image_path:
        result["deed_extraction"] = extract_deed_from_image(image_path)
        extracted_name = result["deed_extraction"].get("buyer_name")
        if extracted_name:
            name = extracted_name
            result["query"] = name

    # Step 2: Search ACRIS for all transactions involving this entity
    parties = search_acris_parties(name)
    seen_docs = set()
    for party in parties:
        doc_id = party.get("document_id")
        if not doc_id or doc_id in seen_docs:
            continue
        seen_docs.add(doc_id)
        if len(seen_docs) > 10:
            break
        doc = get_acris_document(doc_id)
        result["acris_transactions"].append({"document_id": doc_id, "party_name": party.get("name"), "party_type": "buyer" if party.get("party_type") == "2" else "seller", "document_date": doc.get("doc_date"), "document_amount": doc.get("doc_amount"), "doc_type": doc.get("doc_type")})

    # Step 3: Look up the entity in NY Secretary of State records
    dos_results = search_ny_dos(name)
    if dos_results:
        entity = dos_results[0]
        result["dos_entity"] = {"entity_name": entity.get("current_entity_name"), "dos_id": entity.get("dos_id"), "entity_type": entity.get("entity_type"), "filing_date": entity.get("initial_dos_filing_date"), "jurisdiction": entity.get("jurisdiction"), "county": entity.get("county")}

        # Extract the people behind the LLC
        owners = []
        if entity.get("chairman_name"):
            owners.append({"name": entity["chairman_name"], "role": "CEO/Chairman", "address": ", ".join(filter(None, [entity.get("chairman_address_1"), entity.get("chairman_city"), entity.get("chairman_state"), entity.get("chairman_zip")]))})
        if entity.get("dos_process_name"):
            owners.append({"name": entity["dos_process_name"], "role": "Service of Process", "address": ", ".join(filter(None, [entity.get("dos_process_address_1"), entity.get("dos_process_city"), entity.get("dos_process_state"), entity.get("dos_process_zip")]))})
        if entity.get("registered_agent_name"):
            owners.append({"name": entity["registered_agent_name"], "role": "Registered Agent", "address": ", ".join(filter(None, [entity.get("registered_agent_address_1"), entity.get("registered_agent_city"), entity.get("registered_agent_state"), entity.get("registered_agent_zip")]))})
        result["resolved_owners"] = owners

    return result

What does this code do?

This function orchestrates the full pipeline in three stages:

  • If an image path is provided, it calls extract_deed_from_image() to pull out the buyer name. That extracted name then becomes the query for the next two stages.
  • It searches ACRIS for party records matching the name, then fetches the master document for each unique document ID. The seen_docs set prevents duplicate lookups, and the cap of 10 keeps API calls reasonable.
  • It queries the NY DOS dataset and extracts the CEO/Chairman, Service of Process contact, and Registered Agent with their addresses.

Why cap at 10 documents?

Each document ID requires a separate API call to fetch the master record. Large entities can appear in hundreds of transactions. Capping at 10 keeps the resolver responsive while still providing a representative transaction history.

  • Save resolver.py.

Seeing a NameError for Optional?

The Optional type hint is imported from typing at the top of your file. Check that your imports section still includes from typing import Optional.

Help me debug this NameError.

Add the CLI interface

Python's built-in argparse module lets you turn any script into a proper command-line tool with named arguments and help text. You will add a CLI that accepts an LLC name and an optional deed image path.

  • At the very bottom of resolver.py, after the resolve_ownership function, add the following CLI entry point:
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="NYC LLC Ownership Resolver - Pierce the corporate veil")
    parser.add_argument("name", help="LLC or buyer name to resolve")
    parser.add_argument("--image", help="Path to a scanned deed image (JPEG/PNG)", default=None)
    args = parser.parse_args()

    print(f"\n\U0001f50d Resolving ownership for: {args.name}\n")
    result = resolve_ownership(args.name, args.image)
    print(json.dumps(result, indent=2, default=str))

What does this code do?

  • The if __name__ == "__main__": guard ensures this code only runs when you execute the script directly (not when importing it as a module).
  • The name argument is positional (required). The --image flag is optional.
  • The output uses json.dumps with indent=2 for human-readable formatting and default=str to handle any non-serializable values like dates.
  • Save resolver.py.

Indentation error?

Make sure the if __name__ line starts at the leftmost column (zero indentation). It should be at the same level as your function definitions, not nested inside any function.

Help me fix my indentation issue.

Run the full pipeline end-to-end

Time to test the complete resolver. You will pass in a real LLC name and watch all three modules work together.

  • Run the resolver with a test LLC name by running this command:
python resolver.py "EXAMPLE LLC"

What should I see?

You should see a JSON output with five top-level keys: query, deed_extraction, acris_transactions, dos_entity, and resolved_owners.

Some fields may be empty depending on whether that specific LLC appears in ACRIS or the NY DOS dataset. The important thing is that the pipeline runs without errors and returns structured JSON.

  • Try a different entity name that you know operates in NYC real estate to see richer results.
  • If you have a deed image from the previous step, test the image extraction path by running:
python resolver.py "ANY NAME" --image path/to/your/deed.jpg

When using --image, the resolver extracts the buyer name from the deed and uses that as the query for both ACRIS and NY DOS lookups.

Getting a connection error or empty results?

  • Make sure you have an active internet connection. The resolver makes live API calls to NYC Open Data and NY State Open Data.
  • If acris_transactions is empty, try a different name. Not all LLCs appear in ACRIS records.
  • If resolved_owners is empty, the entity may not be registered as a domestic LLC in New York. Try searching for a name you found in the ACRIS results.

Help me troubleshoot my resolver results.

✔️ Awesome, I've got everything!

Great. Make sure you have saved resolver.py with both the resolve_ownership() function and the CLI block.

ⓧ I'd like to double check the full code

Here is the complete resolver.py file at this point:

import anthropic
import requests
import base64
import json
from typing import Optional


ACRIS_PARTIES_ENDPOINT = "https://data.cityofnewyork.us/resource/636b-3b5g.json"
ACRIS_MASTER_ENDPOINT = "https://data.cityofnewyork.us/resource/bnx9-e6tj.json"
NY_DOS_ENDPOINT = "https://data.ny.gov/resource/n9v6-gdp6.json"


def search_acris_parties(name: str, limit: int = 50) -> list:
    """Search ACRIS Real Property Parties by name."""
    safe_name = name.upper().replace("'", "''")
    params = {
        "$where": f"upper(name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
        "$order": "good_through_date DESC",
    }
    response = requests.get(ACRIS_PARTIES_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()


def get_acris_document(document_id: str) -> dict:
    """Get the master record for a given ACRIS document ID."""
    params = {"$where": f"document_id='{document_id}'"}
    response = requests.get(ACRIS_MASTER_ENDPOINT, params=params)
    response.raise_for_status()
    results = response.json()
    return results[0] if results else {}


def search_ny_dos(entity_name: str, limit: int = 10) -> list:
    """Search NY DOS Active Corporations by entity name."""
    safe_name = entity_name.upper().replace("'", "''")
    params = {
        "$where": f"upper(current_entity_name) LIKE '%{safe_name}%'",
        "$limit": str(limit),
    }
    response = requests.get(NY_DOS_ENDPOINT, params=params)
    response.raise_for_status()
    return response.json()


def extract_deed_from_image(image_path: str) -> dict:
    """Extract structured data from a scanned deed image using Claude Vision."""
    client = anthropic.Anthropic()

    with open(image_path, "rb") as f:
        image_data = base64.standard_b64encode(f.read()).decode("utf-8")

    if image_path.lower().endswith(".png"):
        media_type = "image/png"
    elif image_path.lower().endswith(".webp"):
        media_type = "image/webp"
    else:
        media_type = "image/jpeg"

    deed_schema = {
        "type": "object",
        "properties": {
            "buyer_name": {"type": ["string", "null"], "description": "The grantee/buyer name as written on the deed"},
            "seller_name": {"type": ["string", "null"], "description": "The grantor/seller name as written on the deed"},
            "property_address": {"type": ["string", "null"], "description": "The property street address"},
            "borough": {"type": ["string", "null"], "description": "NYC borough (Manhattan, Brooklyn, Queens, Bronx, Staten Island)"},
            "block": {"type": ["string", "null"], "description": "Tax block number"},
            "lot": {"type": ["string", "null"], "description": "Tax lot number"},
            "sale_price": {"type": ["number", "null"], "description": "Sale price in dollars"},
            "document_date": {"type": ["string", "null"], "description": "Date of the deed (YYYY-MM-DD if possible)"},
            "is_llc_buyer": {"type": "boolean", "description": "Whether the buyer name contains LLC, L.L.C., or similar"},
        },
        "required": ["buyer_name", "seller_name", "property_address", "is_llc_buyer"],
    }

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=[{"name": "extract_deed", "description": "Extract structured information from a NYC property deed document.", "input_schema": deed_schema}],
        tool_choice={"type": "tool", "name": "extract_deed"},
        messages=[{"role": "user", "content": [{"type": "image", "source": {"type": "base64", "media_type": media_type, "data": image_data}}, {"type": "text", "text": "Extract all available information from this NYC property deed document. Look for the buyer (grantee), seller (grantor), property address, borough/block/lot, sale price, and date."}]}],
    )

    for block in response.content:
        if block.type == "tool_use":
            return block.input
    return {}


def resolve_ownership(name: str, image_path: Optional[str] = None) -> dict:
    """Full ownership resolution pipeline."""
    result = {
        "query": name,
        "deed_extraction": None,
        "acris_transactions": [],
        "dos_entity": None,
        "resolved_owners": [],
    }

    if image_path:
        result["deed_extraction"] = extract_deed_from_image(image_path)
        extracted_name = result["deed_extraction"].get("buyer_name")
        if extracted_name:
            name = extracted_name
            result["query"] = name

    parties = search_acris_parties(name)
    seen_docs = set()
    for party in parties:
        doc_id = party.get("document_id")
        if not doc_id or doc_id in seen_docs:
            continue
        seen_docs.add(doc_id)
        if len(seen_docs) > 10:
            break
        doc = get_acris_document(doc_id)
        result["acris_transactions"].append({"document_id": doc_id, "party_name": party.get("name"), "party_type": "buyer" if party.get("party_type") == "2" else "seller", "document_date": doc.get("doc_date"), "document_amount": doc.get("doc_amount"), "doc_type": doc.get("doc_type")})

    dos_results = search_ny_dos(name)
    if dos_results:
        entity = dos_results[0]
        result["dos_entity"] = {"entity_name": entity.get("current_entity_name"), "dos_id": entity.get("dos_id"), "entity_type": entity.get("entity_type"), "filing_date": entity.get("initial_dos_filing_date"), "jurisdiction": entity.get("jurisdiction"), "county": entity.get("county")}

        owners = []
        if entity.get("chairman_name"):
            owners.append({"name": entity["chairman_name"], "role": "CEO/Chairman", "address": ", ".join(filter(None, [entity.get("chairman_address_1"), entity.get("chairman_city"), entity.get("chairman_state"), entity.get("chairman_zip")]))})
        if entity.get("dos_process_name"):
            owners.append({"name": entity["dos_process_name"], "role": "Service of Process", "address": ", ".join(filter(None, [entity.get("dos_process_address_1"), entity.get("dos_process_city"), entity.get("dos_process_state"), entity.get("dos_process_zip")]))})
        if entity.get("registered_agent_name"):
            owners.append({"name": entity["registered_agent_name"], "role": "Registered Agent", "address": ", ".join(filter(None, [entity.get("registered_agent_address_1"), entity.get("registered_agent_city"), entity.get("registered_agent_state"), entity.get("registered_agent_zip")]))})
        result["resolved_owners"] = owners

    return result


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="NYC LLC Ownership Resolver - Pierce the corporate veil")
    parser.add_argument("name", help="LLC or buyer name to resolve")
    parser.add_argument("--image", help="Path to a scanned deed image (JPEG/PNG)", default=None)
    args = parser.parse_args()

    print(f"\n\U0001f50d Resolving ownership for: {args.name}\n")
    result = resolve_ownership(args.name, args.image)
    print(json.dumps(result, indent=2, default=str))

Your resolver pipeline is complete. Next up, you will package this tool and deploy it as a serverless API on AWS Lambda so anyone can query it over HTTP.

Deploy to AWS Lambda

Your resolver pipeline works perfectly from the command line. But right now, nobody else can use it unless they have Python installed and access to your machine.

In this step, you'll wrap the resolver in an AWS Lambda function and expose it as an HTTP endpoint via API Gateway. Lambda runs your code on demand and costs nothing when idle. This is the exact architecture real estate intelligence platforms use to serve on-demand lookups.

In this step, get ready to:
  • Create the Lambda handler function that accepts HTTP requests.
  • Package your project with dependencies into a deployment zip.
  • Deploy to AWS Lambda and test the live endpoint.
Create the Lambda handler function

Lambda functions receive events differently from CLI scripts. Instead of reading argparse arguments, a Lambda handler receives a JSON event object from API Gateway containing the HTTP request body.

  • Create a new file called lambda_handler.py in your llc-resolver project folder.
  • Add the following code to lambda_handler.py:
import json
import base64
import tempfile
import os
from resolver import resolve_ownership


def handler(event, context):
    """AWS Lambda handler for the LLC Ownership Resolver."""
    # Parse the JSON body from the API Gateway event
    try:
        body = json.loads(event.get("body", "{}"))
    except (json.JSONDecodeError, TypeError):
        body = event if isinstance(event, dict) else {}

    name = body.get("name")
    image_base64 = body.get("image_base64")

    # Return an error if neither name nor image is provided
    if not name and not image_base64:
        return {
            "statusCode": 400,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(
                {
                    "error": "Provide 'name' (LLC/buyer name) or 'image_base64' (deed scan)"
                }
            ),
        }

    # If an image is provided, decode it to a temp file
    image_path = None
    if image_base64:
        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp.write(base64.b64decode(image_base64))
        tmp.close()
        image_path = tmp.name

    try:
        # Run the full resolver pipeline
        result = resolve_ownership(name or "", image_path)
        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(result, default=str),
        }
    finally:
        # Clean up the temp file if one was created
        if image_path and os.path.exists(image_path):
            os.unlink(image_path)

What does this code do?

The handler() function is the entry point Lambda calls when a request arrives. It accepts two parameters: event (the HTTP request data from API Gateway) and context (Lambda runtime metadata).

The function parses a JSON body looking for name and optional image_base64 fields. If an image is provided, it decodes the base64 data into a temporary file so resolve_ownership() can process it. The finally block ensures the temp file is always cleaned up, even if an error occurs.

  • Save lambda_handler.py.
  • Confirm the file exists by running:
ls lambda_handler.py

You should see lambda_handler.py listed in the output.

Don't see the file?

Make sure you're in the correct project directory. Run pwd to check your current location. You should be inside your llc-resolver folder.

Help me figure out why lambda_handler.py isn't showing up

Create the deployment script

Lambda needs all your Python dependencies bundled into a single zip file alongside your code. You'll create a bash script that automates this packaging and deployment process.

  • Create a new file called deploy.sh in your project folder.
  • Add the following code to deploy.sh:
#!/bin/bash
set -e

echo "📦 Packaging Lambda function..."

rm -rf package/ lambda.zip

pip install -r requirements.txt -t package/ --quiet

cp resolver.py package/
cp lambda_handler.py package/

cd package
zip -r ../lambda.zip . -x "*.pyc" -x "__pycache__/*" --quiet
cd ..

echo "✅ lambda.zip created ($(du -h lambda.zip | cut -f1))"

echo ""
echo "🚀 Deploying to AWS Lambda..."

aws lambda create-function \
  --function-name llc-resolver \
  --runtime python3.12 \
  --handler lambda_handler.handler \
  --zip-file fileb://lambda.zip \
  --role arn:aws:iam::$(aws sts get-caller-identity --query Account --output text):role/lambda-execution-role \
  --timeout 30 \
  --memory-size 256 \
  --environment "Variables={ANTHROPIC_API_KEY=$(grep ANTHROPIC_API_KEY .env | cut -d= -f2)}" \
  2>/dev/null || \
aws lambda update-function-code \
  --function-name llc-resolver \
  --zip-file fileb://lambda.zip

echo "✅ Lambda function deployed!"
echo ""
echo "To test: aws lambda invoke --function-name llc-resolver --payload '{\"body\": \"{\\\"name\\\": \\\"EXAMPLE LLC\\\"}\" }' response.json && cat response.json"

What does this script do?

  • set -e makes the script stop immediately if any command fails.
  • pip install -r requirements.txt -t package/ installs all dependencies into a local package/ directory instead of your system Python.
  • The script copies your source files into package/ and zips everything into lambda.zip.
  • aws lambda create-function creates the Lambda function with the python3.12 runtime, 256MB memory, and a 30-second timeout.
  • If the function already exists, the || fallback runs aws lambda update-function-code to update it instead.
  • The --environment flag reads your ANTHROPIC_API_KEY from a local .env file and passes it to Lambda.
  • Save deploy.sh.
  • Make the script executable by running:
chmod +x deploy.sh
  • Confirm both new files exist by running:
ls

You should see deploy.sh, lambda_handler.py, resolver.py, and requirements.txt in the output.

Permission denied when running chmod?

On Windows, you may not need chmod. You can run the script directly with bash deploy.sh instead.

Help me make deploy.sh executable on my system

Deploy to AWS and test the live endpoint

Before running the deploy script, you need two things: an IAM execution role for Lambda, and your API key stored in a .env file that the script can read.

  • Make sure your .env file contains your Anthropic API key. If you don't have one yet, create it by running:
echo "ANTHROPIC_API_KEY=your-api-key-here" > .env

Replace your-api-key-here with your actual Anthropic API key.

  • Create the IAM execution role that Lambda needs by running this command:
aws iam create-role \
  --role-name lambda-execution-role \
  --assume-role-policy-document '{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole"}]}'
  • Attach the basic Lambda execution policy to this role by running:
aws iam attach-role-policy \
  --role-name lambda-execution-role \
  --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

What is this role?

Every Lambda function needs an IAM execution role that grants it permissions. The trust policy tells AWS that the Lambda service is allowed to assume this role.

The AWSLambdaBasicExecutionRole policy gives your function permission to write logs to CloudWatch.

  • Wait about 10 seconds for the role to propagate, then deploy your Lambda function by running:
./deploy.sh

What should I see?

The script will print packaging progress, then either a success message confirming the Lambda function was created, or a message confirming the code was updated (if the function already existed from a previous run).

  • Test the deployed Lambda function directly by running:
aws lambda invoke \
  --function-name llc-resolver \
  --payload '{"body": "{\"name\": \"EXAMPLE LLC\"}" }' \
  response.json && cat response.json

You should see a JSON response containing the resolver output with acris_transactions and dos_entity fields.

Seeing an access denied or role error?

If you see a role-related error, the IAM role may not have finished propagating. Wait 15-20 seconds and run ./deploy.sh again. AWS IAM changes take a few seconds to become globally consistent.

If you see a credentials error, make sure your AWS CLI is configured correctly by running aws sts get-caller-identity. This should return your account ID.

Help me debug my Lambda deployment error

  • Now create an API Gateway REST API to expose your Lambda as an HTTP endpoint. Open the API Gateway console at https://console.aws.amazon.com/apigateway.
  • Click Create API.
  • Under REST API, click Build.
  • For API name, enter llc-resolver-api.
  • Click Create API.
  • On the Resources page, click Create method.
  • For Method type, select POST.
  • For Integration type, select Lambda Function.
  • Toggle on Lambda proxy integration.
  • For Lambda function, select your region and enter llc-resolver.
  • Click Create method.
  • Click Deploy API.
  • For Stage, select New stage and enter prod as the stage name.
  • Click Deploy.
  • Copy the Invoke URL displayed in the Stage details pane.

What is Lambda proxy integration?

With proxy integration enabled, API Gateway passes the entire HTTP request directly to your Lambda function as the event parameter. This is why your handler() function parses event.get("body").

The response format (statusCode, headers, body) is what API Gateway expects back from a proxy-integrated Lambda.

  • Test your live API endpoint by running this curl command (replace the URL with your Invoke URL):
curl -X POST [[API_GATEWAY_URL="your Invoke URL from API Gateway"]] \
  -H "Content-Type: application/json" \
  -d '{"name": "EXAMPLE LLC"}'

You should see a JSON response with the full ownership profile, including ACRIS transactions and DOS entity data.

Getting a 500 Internal Server Error?

This usually means the Lambda function is timing out or hitting a permissions issue. Check the CloudWatch logs for your function in the AWS Console under CloudWatch then Log groups then /aws/lambda/llc-resolver.

If you see a timeout error, the 30-second timeout may not be enough for the external API calls. You can increase it in the Lambda console under Configuration then General configuration.

Help me debug my 500 error from API Gateway

✔️ Awesome, I've got everything!

Great. Double check you've saved all your files before moving on.

ⓧ I'd like to double check the full code

Here is the complete lambda_handler.py:

import json
import base64
import tempfile
import os
from resolver import resolve_ownership


def handler(event, context):
    """AWS Lambda handler for the LLC Ownership Resolver."""
    try:
        body = json.loads(event.get("body", "{}"))
    except (json.JSONDecodeError, TypeError):
        body = event if isinstance(event, dict) else {}

    name = body.get("name")
    image_base64 = body.get("image_base64")

    if not name and not image_base64:
        return {
            "statusCode": 400,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(
                {
                    "error": "Provide 'name' (LLC/buyer name) or 'image_base64' (deed scan)"
                }
            ),
        }

    image_path = None
    if image_base64:
        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp.write(base64.b64decode(image_base64))
        tmp.close()
        image_path = tmp.name

    try:
        result = resolve_ownership(name or "", image_path)
        return {
            "statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps(result, default=str),
        }
    finally:
        if image_path and os.path.exists(image_path):
            os.unlink(image_path)

Here is the complete deploy.sh:

#!/bin/bash
set -e

echo "📦 Packaging Lambda function..."

rm -rf package/ lambda.zip

pip install -r requirements.txt -t package/ --quiet

cp resolver.py package/
cp lambda_handler.py package/

cd package
zip -r ../lambda.zip . -x "*.pyc" -x "__pycache__/*" --quiet
cd ..

echo "✅ lambda.zip created ($(du -h lambda.zip | cut -f1))"

echo ""
echo "🚀 Deploying to AWS Lambda..."

aws lambda create-function \
  --function-name llc-resolver \
  --runtime python3.12 \
  --handler lambda_handler.handler \
  --zip-file fileb://lambda.zip \
  --role arn:aws:iam::$(aws sts get-caller-identity --query Account --output text):role/lambda-execution-role \
  --timeout 30 \
  --memory-size 256 \
  --environment "Variables={ANTHROPIC_API_KEY=$(grep ANTHROPIC_API_KEY .env | cut -d= -f2)}" \
  2>/dev/null || \
aws lambda update-function-code \
  --function-name llc-resolver \
  --zip-file fileb://lambda.zip

echo "✅ Lambda function deployed!"
echo ""
echo "To test: aws lambda invoke --function-name llc-resolver --payload '{\"body\": \"{\\\"name\\\": \\\"EXAMPLE LLC\\\"}\" }' response.json && cat response.json"

Your LLC ownership resolver is now live as a serverless API. It costs exactly $0 when nobody is calling it, and scales automatically when requests come in.

Secret mission

Add Comparable Sales Calculation

Your resolver already identifies who owns an LLC and what deals they've done. But how do you know if they overpaid? In this secret mission, you'll add a market comps feature that pulls recent DEED transactions from the same tax block and calculates whether the LLC's purchase was above or below the neighborhood median.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

This project uses AWS Lambda and API Gateway, both of which are purely pay-per-use. When nobody calls your endpoint, you pay nothing.

Resources you used:

  • AWS Lambda function (llc-resolver).
  • API Gateway REST API.
  • IAM execution role (lambda-execution-role).
  • Local project files and virtual environment.
  • Anthropic API key (stored in your .env file).

Keep everything running

No action needed. Choose this if you want to keep your resolver available as a live API endpoint.

  • Lambda and API Gateway cost $0 when idle. You only pay when someone calls the endpoint.
  • Your Anthropic API key is only charged when you make Claude API calls.
  • The IAM role and local files take up negligible space and cost nothing.

Set a billing alarm for peace of mind

In the AWS Console, navigate to Billing, then Budgets. Create a $5/month budget with email alerts so you are never surprised by unexpected charges.

Pause - I'll come back to this later

Since Lambda and API Gateway are serverless, there is nothing to "pause" in the traditional sense. They already cost $0 when idle.

  • Your endpoint remains live but costs nothing until it receives traffic.
  • If you want to prevent any accidental calls, you can delete the API Gateway stage (the endpoint URL stops working) while keeping the Lambda function intact for later.
  • Your local files remain on your machine ready for future development.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below.

  • Delete the Lambda function by running this command:
  • Delete the API Gateway REST API by opening the API Gateway console at https://console.aws.amazon.com/apigateway. On the APIs page, select your API. Choose API actions, then Delete API, and confirm your choice.
  • Delete the IAM execution role. First detach its policies, then delete the role:
  • Delete your local project files:
aws lambda delete-function --function-name llc-resolver

Next, detach the managed policy from the IAM role and delete the role:

aws iam detach-role-policy --role-name lambda-execution-role --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
aws iam delete-role --role-name lambda-execution-role

Finally, remove your local project files. On macOS or Linux:

cd ~/Desktop
rm -rf llc-resolver

On Windows:

cd ~\Desktop
rmdir /s /q llc-resolver

Nice Work!

Nice Work!

You just built an intelligence tool that does what $100M deal teams do manually. Given any LLC name from a NYC property record, your tool reveals the real humans behind it, their full transaction history, and market context.

You've learned how to:

  • Built a cross-referencing ownership resolver that queries NYC's ACRIS property records and the NY Secretary of State Socrata API to pierce the corporate veil on LLC transactions.
  • Built a Claude Vision-powered deed scanner that extracts structured data from scanned document images using the tool_use pattern for guaranteed JSON output.
  • Deployed the full pipeline as a serverless API on AWS Lambda with API Gateway that costs $0 when idle.
  • Added a comparable sales calculation that estimates property value based on recent ACRIS transactions in the same block.

Ready to quiz yourself?