diff --git a/index.toml b/index.toml index 88948ba..38bc9c8 100644 --- a/index.toml +++ b/index.toml @@ -371,6 +371,12 @@ notebook = "prior_labs_agent.ipynb" new = true topics = ["Agents", "MCP", "Data Processing"] +[[cookbook]] +title = "Document Processing with Unstructured Transform MCP" +notebook = "unstructured_transform_mcp.ipynb" +new = true +topics = ["Agents", "MCP", "Data Processing"] + [[cookbook]] title = "Agentic Itinerary Planning with OpenStreetMap" notebook = "agentic_itinerary_planning_openstreetmap.ipynb" diff --git a/notebooks/unstructured_transform_mcp.ipynb b/notebooks/unstructured_transform_mcp.ipynb new file mode 100644 index 0000000..97f170d --- /dev/null +++ b/notebooks/unstructured_transform_mcp.ipynb @@ -0,0 +1,225 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "t1a2b3c4", + "metadata": {}, + "source": [ + "# Document Processing with Unstructured Transform MCP\n", + "\n", + "In this recipe, we connect to [Unstructured Transform](https://docs.unstructured.io/transform/overview)'s hosted MCP server and use a Haystack `Agent` to parse and chunk a document, entirely through MCP tools.\n", + "\n", + "**Services used:**\n", + "- [Unstructured Transform MCP](https://mcp.transform.unstructured.io): document processing (partition, enrich, chunk, embed) exposed as MCP tools\n", + "- [Anthropic Claude](https://www.anthropic.com/): LLM for agent reasoning" + ] + }, + { + "cell_type": "markdown", + "id": "i1nst4ll0", + "metadata": {}, + "source": [ + "## Install dependencies" + ] + }, + { + "cell_type": "code", + "id": "dep5nd001", + "metadata": {}, + "source": [ + "!pip install -q haystack-ai mcp-haystack anthropic-haystack" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "4p1k3ys01", + "metadata": {}, + "source": [ + "## Set up API keys\n", + "\n", + "You'll need two API keys:\n", + "- **Unstructured API key**: get one from the [Transform get-started page](https://transform.unstructured.io/get-started) after signing in. The free tier includes 15,000 pages a month.\n", + "- **Anthropic API key**: get one at [console.anthropic.com](https://console.anthropic.com/)" + ] + }, + { + "cell_type": "code", + "id": "k3ys00cod", + "metadata": {}, + "source": [ + "import os\n", + "from getpass import getpass\n", + "\n", + "if \"UNSTRUCTURED_API_KEY\" not in os.environ:\n", + " os.environ[\"UNSTRUCTURED_API_KEY\"] = getpass(\"Enter your Unstructured API key: \")\n", + "if \"ANTHROPIC_API_KEY\" not in os.environ:\n", + " os.environ[\"ANTHROPIC_API_KEY\"] = getpass(\"Enter your Anthropic API key: \")" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "st3p1md01", + "metadata": {}, + "source": [ + "## Step 1: Connect to Unstructured Transform MCP\n", + "\n", + "[Unstructured Transform](https://docs.unstructured.io/transform/overview) exposes its document-processing pipeline (partition, enrich, chunk, embed) as a hosted MCP server at `https://mcp.transform.unstructured.io`. We connect to it with [`MCPToolset`](https://docs.haystack.deepset.ai/docs/mcptoolset), using `StreamableHttpServerInfo`'s native `token` parameter to send the Unstructured API key as an `Authorization: Bearer` header." + ] + }, + { + "cell_type": "code", + "id": "st3p1cod1", + "metadata": {}, + "source": [ + "from haystack_integrations.tools.mcp import MCPToolset, StreamableHttpServerInfo\n", + "from haystack.utils import Secret\n", + "\n", + "server_info = StreamableHttpServerInfo(\n", + " url=\"https://mcp.transform.unstructured.io\",\n", + " token=Secret.from_env_var(\"UNSTRUCTURED_API_KEY\"),\n", + ")\n", + "toolset = MCPToolset(server_info=server_info, eager_connect=True)\n", + "\n", + "for tool in toolset.tools:\n", + " print(f\"{tool.name}: {tool.description}\")" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "st3p1md02", + "metadata": {}, + "source": [ + "The pipeline runs asynchronously as a job: submit a file for processing, poll until it's done, then fetch the rendered result; a separate helper mints an upload URL for files that aren't already reachable over HTTPS. Unstructured adds tools and capabilities to this server as they ship new features, so rather than list exact tool names and a fixed count here, the cell above discovered the live toolset at connect time, and the agent below matches tools to each step by description." + ] + }, + { + "cell_type": "markdown", + "id": "st3p2md01", + "metadata": {}, + "source": [ + "## Step 2: Build a Haystack Agent with the Transform MCP toolset\n", + "\n", + "We give the agent the toolset directly, along with a system prompt describing the asynchronous submit -> poll -> fetch flow by behavior rather than by hardcoded tool name, since the agent needs to poll for a result rather than get one back immediately, and this way the prompt keeps working as Unstructured renames or adds tools." + ] + }, + { + "cell_type": "code", + "id": "st3p2cod1", + "metadata": {}, + "source": [ + "import time\n", + "from typing import Annotated\n", + "\n", + "from haystack.components.agents import Agent\n", + "from haystack.tools import tool\n", + "from haystack_integrations.components.generators.anthropic import AnthropicChatGenerator\n", + "\n", + "\n", + "@tool\n", + "def wait(seconds: Annotated[int, \"How many seconds to pause before the next tool call\"]) -> str:\n", + " \"\"\"Pause execution for the given number of seconds. A chat generator can't control timing on\n", + " its own, so call this between status checks instead of just waiting in the response text.\"\"\"\n", + " time.sleep(seconds)\n", + " return f\"Waited {seconds} second(s).\"\n", + "\n", + "\n", + "agent = Agent(\n", + " chat_generator=AnthropicChatGenerator(\n", + " model=\"claude-opus-4-6\",\n", + " generation_kwargs={\"max_tokens\": 4096},\n", + " ),\n", + " tools=toolset + wait,\n", + " system_prompt=\"\"\"You are a document-processing assistant with access to Unstructured Transform MCP tools. Check the tools available to you and use whichever ones match the steps below by description, since exact tool names may change over time.\n", + "\n", + "Transform jobs are asynchronous. When asked to process a document:\n", + "1. Submit the file reference(s) and the requested processing stages to start a processing job. This returns a job ID immediately; the job itself runs in the background.\n", + "2. Check the job's status. If it isn't complete yet, call the wait tool for a few seconds, then check again, repeating until it reports as complete.\n", + "3. Fetch the job's rendered output using its job ID, and summarize it for the user.\n", + "\"\"\",\n", + ")" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "st3p3md01", + "metadata": {}, + "source": [ + "## Step 3: Process a document end-to-end\n", + "\n", + "We hand the agent a publicly reachable PDF and ask it to parse and chunk it. No local file or upload step is needed here, since the job-submission tool accepts `https://` URLs directly." + ] + }, + { + "cell_type": "code", + "id": "st3p3cod1", + "metadata": {}, + "source": [ + "from haystack.dataclasses import ChatMessage\n", + "\n", + "pdf_url = \"https://arxiv.org/pdf/1706.03762\"\n", + "\n", + "result = agent.run(\n", + " messages=[\n", + " ChatMessage.from_user(\n", + " f\"Parse and chunk the PDF at {pdf_url}. \"\n", + " \"Use the 'hi_res' partition strategy, and chunk with chunk_by_title, \"\n", + " \"max_characters=1000. Once the job is complete, fetch the results as \"\n", + " \"markdown and show me the first two chunks.\"\n", + " )\n", + " ]\n", + ")" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "code", + "id": "st3p3cod2", + "metadata": {}, + "source": [ + "print(result[\"last_message\"].text)" + ], + "outputs": [], + "execution_count": null + }, + { + "cell_type": "markdown", + "id": "c0ncl0001", + "metadata": {}, + "source": [ + "## Conclusion\n", + "\n", + "Unstructured Transform's MCP server brings partitioning, enrichment, chunking, and embedding into a single set of tools an agent can call directly, without wiring up a separate ETL pipeline. Because job submission runs asynchronously and returns a job ID right away, a Haystack `Agent` can poll for completion and fetch results the same way it would call any other tool, making it straightforward to drop document processing into a larger agentic workflow." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "3.12.12", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.12" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +}