Full Doc Agentic Search of Large OCR Documents in Amazon Quick

Full-document agentic search for large OCR files in Quick - leveraging MCP tools and the code interpreter to circumvent timeout and character-limit constraints in scenarios where RAG is not an acceptable approach .

Srinivasa amirapu

Published Jun 23, 2026

Last modified Jul 13, 2026

Co-Author: Brett Galloway

Background

Amazon Quick is an AI-powered assistant that combines conversational AI with tool-use capabilities via the Model Context Protocol (MCP). When a user asks a question, the Quick Orchestrator reasons about the query, invokes Bedrock models, and can call external MCP tools to retrieve data, process documents, or perform actions.

Regulated industries generate documents that are:

· Large: 200–500+ pages upto 5 MB extracted text

· Image-heavy: scanned filings, architectural diagrams, handwritten forms, tables as images

· Unstructured: no searchable text layer; OCR is the only path to content extraction

· Compliance-critical: every page must be verifiably processed; skipping pages is a regulatory violation which could take long time processing > MCP/ALB time limit.

These documents require full document agentic search (RAG not acceptable) with rasterization (PDF page → high-DPI image), OCR (image → text via vision models), and multi-agent analysis (L1 orchestrator → L2 specialists).

Why Large Documents Must Go Through S3 + Code Interpreter

A natural question: Why not just upload the PDF directly to the Quick Agent and let it handle the content natively?

Like any agentic product, native document handling has known limitations with large, image-heavy files. The key constraints are:

Limitation	Impact on Large Docs
File size caps	Any Agentic Product's native upload has size limits; 300-page scanned PDFs exceed them
Context window limit	Even if text were extractable, it could exceed model context window
Character count limit	Even if text were extractable, it could exceed current character count limit
MCP/ALB Timeout Limit	Some documents could take > MCP/ALB Timeout Limit to process for full doc agent search prompts

The workaround: Use Quick’s Code Interpreter available to push the document to S3, then invoke an MCP Server that specializes in document processing:

User uploads 300-page PDF → Code Interpreter execute python code to upload it to S3 using pre-signed url
↓
Quick Orchestrator → MCP tool action (Process OCR/PDF)
↓
MCP Server fetches from S3 → rasterizes → vision model OCR → returns extracted content

This flow visible in the architecture diagram. The MCP Server is purpose-built for the heavy lifting for large documents (Any document hitting one of file size caps , Content window limit , Character count limit , MCP/ALB Timeout Limit).

Figure 1: Chat Assistant architecture. User interacts via Quick UI. The Quick Orchestrator invokes Bedrock models for reasoning and issues tool actions (Process OCR/PDF) to the MCP Server for document processing. Code Interpreter pushes large documents to S3; the MCP Server fetches and processes them.

Component Roles:

Component	Role
Quick UI	User-facing chat interface
Quick Orchestrator	Reasoning engine — decides when to invoke models vs. tools
Bedrock Model	LLM for reasoning, summarization, and vision-based OCR
Code Interpreter	Executes python code than can push uploaded documents to S3 (workaround for native size/format limits)
S3	Durable storage for documents the MCP Server processes
MCP Server (AgentCore)	Hosts OCR/analysis tools — rasterizes pages, runs vision models, returns structured content
Quick Memory	Stores conversation history and user preferences
IDC	Identity Center for user authentication

The Solution: MCP Tasks Extension

Client calls tool → Server returns immediately with a taskId + status: "working"
Client polls → Calls get_task_status repeatedly until status: "completed"
Each poll is fast → No single request exceeds the 60-second timeout
Progress is visible → Intermediate statusMessage shows what stage the work is in

Why DynamoDB as the Task Store

AgentCore Runtime is stateless by design: any request - including the poll that follows a tool call — can land on a different container instance than the one that started the task. An in-process store (a dict, a local file) would make the task invisible to any instance other than the one that created it, so a client's get_task_status call would non-deterministically succeed or fail depending on which
instance happened to handle it.

This rules out anything tied to a single container and points to an external,shared store. Among the common options - S3, ElastiCache/Redis, RDS, DynamoDB.

DynamoDB fits this workload specifically:

No connection management. Unlike RDS, there's no connection pool to size or warm up, which matters when container instances scale up and down frequently.
Atomic per-item updates. UpdateItem lets the background thread write statusMessage and the final result/error without a read-modify-write race against a client polling the same item concurrently.
Single-digit-millisecond reads. Since get_task_status may be called repeatedly and in parallel across many pages of a single document, read latency directly adds up across a full-document scan
DynamoDB keeps each individual poll cheap.
Scales to zero and back automatically. No cluster to provision or pay for between bursts of document-processing traffic, unlike Redis/ElastiCache.

Implementation

1. Pre-signed url mcp tool

Generate a presigned S3 upload URL for client side upload using quick's code interpreter.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@mcp.tool()
def get_presigned_upload_url(filename: str, bucket: str = "doc-ocr-mcp-619071352830-us-east-1", expires_in_seconds: int = 3600) -> dict:
    """Generate a presigned S3 upload URL for client side upload using code interpreter.

    Args:
        filename: Name for the file in S3 (stored under uploads/<filename>).
        bucket: S3 bucket name.
        expires_in_seconds: URL validity window (default: 3600 = 1 hour).

    Returns:
        Presigned URL, upload status, and S3 URI.
    """
    import http.client

    import boto3

    s3 = boto3.client("s3")
    key = f"uploads/{filename}"
    url = s3.generate_presigned_url(
        ClientMethod="put_object",
        Params={"Bucket": bucket, "Key": key},
        ExpiresIn=expires_in_seconds,
        HttpMethod="PUT",
    )

   
    # return URL for client-side upload using code interpreter
    return {
        "content": [{"type": "text", "text": url}],
        "metadata": {
            "s3_uri": f"s3://{bucket}/{key}",
            "presigned_url": url,
            "http_method": "PUT",
            "expires_in_seconds": expires_in_seconds,
            "usage_curl": f"curl -X PUT --upload-file <local-path> '{url}'",
        },
    }

2. DynamoDB Task Store

A shared task store. Must be readable/writable from any container instance, since AgentCore Runtime (and most container platforms) can route your next request to a different instance than the one that started the task. Use DynamoDB (or any shared, low-latency KV store) — do not use in-memory dicts or local files.

Minimum schema: taskId (partition key), status, createdAt, lastUpdatedAt, plus whatever result/error/progress fields your domain needs.

_ddb = boto3.resource("dynamodb", region_name=os.environ.get("AWS_REGION", "us-east-1"))
_task_table = _ddb.Table(os.environ.get("TASK_TABLE", "<your-table-name>"))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import threading, uuid, json, os
from datetime import datetime, timezone
import boto3

_TASK_TABLE = os.environ.get("TASK_TABLE", "doc-ocr-mcp-tasks")
_ddb = boto3.resource("dynamodb", region_name="us-east-1")
_task_table = _ddb.Table(_TASK_TABLE)

def _create_task() -\> str:
    """Create a task record. Returns task_id."""
    task_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc).isoformat()
    _task_table.put_item(Item={
        "taskId": task_id,
        "status": "working",
        "createdAt": now,
        "lastUpdatedAt": now,
    })
    return task_id

def _update_task_progress(task_id: str, step: str) -\> None:
    """Write intermediate status for client polling."""
    _task_table.update_item(
        Key={"taskId": task_id},
        UpdateExpression="SET statusMessage = :step, lastUpdatedAt = :t",
        ExpressionAttributeValues={
            ":step": step,
            ":t": datetime.now(timezone.utc).isoformat(),
        },
    )

def _complete_task(task_id: str, result: dict) -\> None:
    """Mark task as completed with result payload."""
    _task_table.update_item(
        Key={"taskId": task_id},
        UpdateExpression="SET #s = :s, #r = :r, lastUpdatedAt = :t",
        ExpressionAttributeNames={"#s": "status", "#r": "result"},
        ExpressionAttributeValues={
            ":s": "completed",
            ":r": json.dumps(result),
            ":t": datetime.now(timezone.utc).isoformat(),
        },
    )

def _fail_task(task_id: str, error: str) -\> None:
    """Mark task as failed with error message."""
    _task_table.update_item(
        Key={"taskId": task_id},
        UpdateExpression="SET #s = :s, #e = :e, lastUpdatedAt = :t",
        ExpressionAttributeNames={"#s": "status", "#e": "error"},
        ExpressionAttributeValues={
            ":s": "failed",
            ":e": error,
            ":t": datetime.now(timezone.utc).isoformat(),
        },
    )

3. Long Running Tool call that returns task id immediately

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@mcp.tool()

def start_my_work(input_a: str, input_b: int) -> dict:

    task_id = _create_task()  # writes {taskId, status: "working", createdAt, lastUpdatedAt}



    def _run():

        try:

            _update_task_progress(task_id, "step 1 description")

            # ... do step 1 ...

            _update_task_progress(task_id, "step 2 description")

            # ... do step 2 ...

            result = { ... }

            _complete_task(task_id, result)

        except Exception as e:

            _fail_task(task_id, str(e))



    threading.Thread(target=_run, daemon=True).start()



    return {

        "content": [{"type": "text", "text": json.dumps({

            "taskId": task_id,

            "status": "working",

            "message": "Work started. Poll get_task_status(task_id) for result.",

        })}],

    }

Key Points :- Create the task record before spawning the thread, so a poll immediately after the tool call always finds a record. - Use a daemon thread so it doesn’t block container shutdown, but understand the tradeoff: if the container is recycled mid-task, the task is lost. For work that must survive container recycling, use a real queue (SQS + a worker fleet, or Step Functions) instead of an in-process thread. Use threads only when “best effort, and mostly-long-lived containers” is an acceptable model. - Wrap the entire background function body in try/except and always call _fail_task on exception — an unhandled exception in the thread will otherwise leave the task stuck in "working" forever with no error surfaced.

Keep below three states and field names across your org’s MCP servers so client-side polling logic can be shared:

status	required fields	notes
working	taskId, status, statusMessage, lastUpdatedAt	optionally progressTrace
completed	(your tool’s normal successful-result shape)	no status field needed if the shape is unambiguous
failed	taskId, status, error	error should be a string

4. Polling Tool — get_task_status

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@mcp.tool()

def get_task_status(task_id: str) -> dict:

    resp = _task_table.get_item(Key={"taskId": task_id})

    task = resp.get("Item")



    if not task:

        return {"content": [{"type": "text", "text": json.dumps({

            "error": "Task not found", "taskId": task_id,

        })}]}



    if task["status"] == "completed":

        return json.loads(task["result"])  # return the final result verbatim



    if task["status"] == "failed":

        return {"content": [{"type": "text", "text": json.dumps({

            "taskId": task_id, "status": "failed", "error": task.get("error", "unknown"),

        })}]}



    # still working

    return {"content": [{"type": "text", "text": json.dumps({

        "taskId": task_id,

        "status": "working",

        "statusMessage": task.get("statusMessage", ""),

        "lastUpdatedAt": task.get("lastUpdatedAt"),

    })}]}

Key points: - On "completed", return the stored final result, not a wrapper around it — the caller should get the same shape whether they got the result from get_task_status or a synchronous equivalent tool. This lets callers treat “poll until done” as transparent. If you want to influence how often the caller polls, communicate that via a field in the response (e.g. pollIntervalMs) and let the caller’s own loop/scheduler sleep — don’t block the MCP request thread server-side. Update statusMessage (and optionally a structured progressTrace) from inside the background thread at each meaningful step, so polling clients can show real progress rather than a static “working” spinner.

Decision Framework

Signal	Use MCP Tools	Quick Native
Document size	> 20 OCR pages	Small documents
>Character Count limit	Upload to S3 using code interpreter and Make MCP tools to process the doc	<= Character Count Limit
Compliance requirement	Full-document scan mandatory	Selective retrieval OK

Production Hardening

DynamoDB TTL — Set expiresAt attribute with a 1-hour offset. Auto-delete stale tasks.
Idempotency — Cache by (pdf_hash, page_number, question_hash) to avoid reprocessing.
Dead letter handling — Sweeper Lambda marks stale "working" tasks as failed after 15 minutes.
Observability — CloudWatch metrics: TaskCreated, TaskCompleted, TaskFailed, TaskDurationMs. Alarm on failure rate.

IAM Policy (Minimum Privilege)

1
2
3
4
5
6
7
8
9
10
11
aws iam put-role-policy \
  --role-name \<AgentCore-execution-role\> \
  --policy-name doc-ocr-mcp-dynamodb \
  --policy-document '{
    "Version": "2012-10-17",
    "Statement": [{
      "Effect": "Allow",
      "Action": ["dynamodb:PutItem","dynamodb:GetItem","dynamodb:UpdateItem"],
      "Resource": "arn:aws:dynamodb:us-east-1:ACCOUNT:table/doc-ocr-mcp-tasks"
    }]
  }'

What's Next

Near-Term: Native Rasterization & OCR in Quick Orchestrator

The ideal end-state eliminates the S3 + MCP Server roundtrip entirely:

Current State	Future State
User uploads PDF → Code Interpreter → S3 → MCP Server rasterizes → vision model → return	User uploads PDF → Quick Orchestrator natively rasterizes pages → built-in vision OCR → direct response

What "native support" means:

Orchestrator handles PDF rasterization internally (page → image conversion)
Built-in OCR pipeline using Bedrock vision models — no external MCP call needed
Per-page streaming results — user sees content as each page completes

Until native rasterization ships, the MCP Server + async tasks pattern is the production-validated approach for large document processing.

References

# agentic-ai # amazon-quick-suite

Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.

Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article

Builder Center

Full Doc Agentic Search of Large OCR Documents in Amazon Quick

Srinivasa amirapu

Background

Why Large Documents Must Go Through S3 + Code Interpreter

The Solution: MCP Tasks Extension

Why DynamoDB as the Task Store

Implementation

1. Pre-signed url mcp tool

2. DynamoDB Task Store

3. Long Running Tool call that returns task id immediately

4. Polling Tool — get_task_status

Decision Framework

Production Hardening

IAM Policy (Minimum Privilege)

What's Next

Near-Term: Native Rasterization & OCR in Quick Orchestrator

References