LangChain iGPT

LangChain integration for the iGPT Personal Data Retrieval API

• Website: https://www.igpt.ai
• Documentation: https://docs.igpt.ai
• Playground: https://igpt.ai/hub/playground/

Requirements

• Python >= 3.10

Installation

pip install -U langchain-igpt

Set these environment variables (recommended):

export IGPT_API_KEY="your-api-key"
export IGPT_API_USER="your-api-user"

Alternatively, pass api_key and user directly to tool constructors.

Connecting Datasources

If you receive warnings like “No datasources found”, it means the user has not connected any datasources yet. Authorize and connect datasources here.

IgptAsk tool

Generate a response based on the input and related context. ↗

Parameters

• input (string, required): The prompt/question to ask.

• user (string, optional if set in constructor): Unique user identifier.

• stream (boolean, optional, default: false): If true, returns iterable stream.

• quality (string, optional): Context engineering quality (default: "cef-1-normal"). Read more.

• output_format (string | object, optional):

  • "text" (default)
  • "json"
  • { schema: <JSON Schema> } to enforce a structured output

NOTE:

• If you set user, stream, quality, or output_format during invocation, that value will overwrite the value passed during instantiation.

Import & Initialize

from langchain_igpt import IgptAsk

tool = IgptAsk(user="user_123", api_key="ak:...", quality="cef-1-normal")

Invoke directly

res = tool.invoke({
    "input": "Summarize my last meeting in 5 bullet points."
})
print(res)

Example response:

{
  "id": "...",
  "output": "...",
  "context": {
    "quality": "cef-1-normal",
    "indexed": 0.0001783,
    "datasources": [{...}]
  },
  "metadata": {
    "sources": [{...}]
  },
  "usage": {
    "input_tokens": 1272,
    "output_tokens": 281,
    "total_tokens": 1553
  }
}

Invoke directly with JSON Schema

Use a schema to get consistent, machine-validated structure.

output_format = {
  "strict": True,
  "schema": {
      "type": "object",
      "properties": {
        "action_items": {
          "type": "array",
          "description": "List of action items",
          "items": {
            "type": "object",
            "properties": {
              "title": { "type": "string", "description": "Short summary of the action item" },
              "owner": { "type": "string", "description": "Person responsible for the action item" },
              "due_date": { "type": "string", "format": "date", "description": "Expected completion date" }
            },
            "required": ["title", "owner", "due_date"],
            "additionalProperties": False
          }
        }
      },
      "required": ["action_items"],
      "additionalProperties": False
  }
}
res = tool.invoke({
    "input": "Extract all action items from yesterday’s board meeting.",
    "output_format": output_format
})
print(res)

Example response (schema):

{
  "action_items": [
    {
      "title": "Approve revised Q1 budget allocation",
      "owner": "Board of Directors",
      "due_date": "2026-01-15"
    },
    {
      "title": "Approve final FY2026 strategic priorities",
      "owner": "Board of Directors",
      "due_date": "2026-01-31"
    }
  ]
}

Invoke asynchronously

async def main():
    result = await tool.ainvoke({"input": "List all vendor escalations from last month"})
    print(result)
asyncio.run(main())

Streaming IgptAsk

For streaming responses, set stream: True. The tool returns an iterable that yields parsed JSON chunks.

for chunk in tool.invoke({"input":"Summarize yesterday’s meeting and list action items.", "stream":True, "quality":"cef-1-normal", "output_format": "text"}):
    print(chunk, end="", flush=True)

Using IgptAsk inside a LangChain Agent

This example shows how to plug the IgptAsk tool into a LangChain agent, so the LLM can decide when to call the tool to retrieve grounded context and then compose an answer for the user.

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_igpt import IgptAsk

ask_tool = IgptAsk()

agent = create_agent(
    model=ChatOpenAI(model="gpt-5", api_key="sk-..."),
    tools=[ask_tool],
    system_prompt=(
        "You are a helpful research assistant. Your primary job is to retrieve, "
        "summarize, and interpret user data via the igpt_ask tool."
    )
)

response = agent.invoke({
    "messages": [
        {"role": "user", "content": "Summarize key risks, decisions, and next steps from this week's meetings."}
    ]
})
print(response)

Streaming responses: Step-by-step updates (streams state updates for each agent step)

for chunk in agent.stream(
    {
        "messages": [{"role": "user", "content": "Summarize the last 5 emails and highlight any actions I need to take."}]
    },
    stream_mode="updates"
):
    for step, data in chunk.items():
        print(f"step: {step}")
        print(f"content: {data['messages'][-1].content_blocks}")

Streaming responses: Token-by-token streaming

for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "Summarize PTO approvals"}]},
    stream_mode="messages",
):
    print(f"node: {metadata['langgraph_node']}")
    print(f"content: {token.content_blocks}")
    print("\n")

IgptSearch tool

Search in connected datasources. ↗

Parameters

• query (string, optional): Search query to execute.
• user (string, optional if set in constructor): Unique user identifier.
• date_from (string, optional): Start date filter (YYYY-MM-DD).
• date_to (string, optional): End date filter (YYYY-MM-DD).
• max_results (number, optional): Limit number of results (e.g., 50).

NOTE:

• query, date_from, and date_to are intended to be provided during invocation (i.e., per-request), not as fixed constructor defaults. This lets the agent compute them dynamically (e.g., "last 30 days" -> concrete YYYY-MM-DD dates).
• If you set user or max_results during invocation, that value will overwrite the value passed during instantiation.

Import & Initialize

from langchain_igpt import IgptSearch

search_tool = IgptSearch(
    user="user_123",
    api_key="ak-...",
    max_results=5
)

Invoke directly

res = search_tool.invoke({"query": "invoices", "date_from":"2025-01-02"})
print(res)

Example response:

{
  "id": "...",
  "context": {
    "indexed": 1,
    "datasources": [{...}]
  },
  "results": [
    {
      "id": "...",
      "timestamp": "2025-05-13T11:48:34Z",
      "type": "message",
      "from": "...",
      "content": "..."
    }
  ],
  "durations": {
    "retrieve": 47,
    "resolve": 10
  }
}

Invoke asynchronously

async def main():
    result = await search_tool.ainvoke({"query": "meeting notes"})
    print(result)
asyncio.run(main())

Using IgptSearch inside a LangChain agent

This example shows how to use the IgptSearch tool inside a LangChain agent.

What happens when you run it:

• The agent receives the user's request: "Find invoices/receipts from the last 30 days."
• The agent uses the LLM (ChatOpenAI) to translate the request into an IgptSearch tool call. Typically, the LLM will:
  • set query to something like "invoice OR receipt"
  • set date_from to today minus 30 days and date_to to today, formatted as YYYY-MM-DD
• IgptSearch queries the user’s connected datasources and returns raw results + metadata.

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_igpt import IgptSearch

search_tool = IgptSearch()
agent = create_agent(
    model=ChatOpenAI(model="gpt-5", api_key="sk-..."),
    tools=[search_tool],
    system_prompt=(
        "You are a research assistant that retrieves user data using the igpt_search tool."
    )
)
response = agent.invoke({
    "messages": [
        {"role": "user", "content": "Find invoices/receipts from the last 30 days."}
    ]
})
print(response)

IgptSearchRetriever

IgptSearchRetriever is a LangChain-compatible retriever that provides search over connected personal datasources via the iGPT Search API.

Import & Initialize

from langchain_igpt import IgptSearchRetriever

retriever = IgptSearchRetriever(max_results=50)

Usage

docs = retriever.invoke("contract renewal date", date_from="2025-01-02")
print(docs)

[Document(metadata={'id': '<ID>', 'timestamp': '2026-01-22T14:40:12Z', 'type': 'message', 'from': '<FROM>', 'subject': 'Subscription renewal reminder Dec 17, 2025', 'source': '<SOURCE>'}, page_content='...')]

Async Usage

async def main():
    docs = await retriever.ainvoke("contract renewal date")
    print(docs)

asyncio.run(main())

Use within a chain

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_igpt import IgptSearchRetriever

retriever = IgptSearchRetriever()

prompt = ChatPromptTemplate.from_template("""
Answer the question using only the context below.

Context:
{context}

Question:
{question}
""")

llm = ChatOpenAI(model="gpt-5", api_key="sk-...")

def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | llm
    | StrOutputParser()
)

res = chain.invoke("Find Invoice/Receipt")
print(res)

('Here are all invoices/receipts found in the provided context:\n'
 '\n'
 '- Your Company (receipt)\n'
 '  - Invoice ID: ...\n'
 '  - Team: Product+Marketing\n'
 '  - Period: Jan 17, 2026 – Feb 17, 2026\n'
 '  - Total: $100.00 USD\n'
 '\n'
 '- Your Company (Invoice Receipt)\n'
 '  - Invoice Number: ...\n'
 '  - Invoice Date: 2025-11-05\n'
 '  - Customer: Jane Doe\n'
 '  - Product: Developer Membership\n'
 '  - Total: $99.00 USD (paid via credit card)\n'
 '\n'
 '- Your Business Name (WHMCS-style invoice)\n'
 '  - Invoice #: ...\n'
 '  - Date: 06/23/2018\n'
 '  - Invoiced To: John Smith\n'
 '  - Total: $37.50 USD\n'
 '  - Transaction ID: ...')

License

MIT