I Built a Brex Connector for Cognee: Bringing Business Expense Data Into AI Memory 🚀

What I learned building an open-source data connector during MergeTober by WeMakeDevs.

What if you could ask an AI assistant about business expenses instead of manually searching through transaction records, notes, and budgets?

That question captures the idea behind my recent open-source contribution to Cognee Community.

During MergeTober by WeMakeDevs, I worked on building a Brex data-source connector for Cognee. The work involved much more than fetching data from an API. I also had to think about how the data should be organized, how future syncs should work, and how to protect existing data when something goes wrong.

Here is what I built, how it works, and what I learned.

🌍 My entry into MergeTober

Open source can feel intimidating when you're starting out. You might wonder which project to choose, how to understand an unfamiliar codebase, or what maintainers expect from a contribution.

MergeTober gave me an opportunity to work on a real integration issue rather than build another standalone demo.

My workflow was straightforward:

1. Find an open issue that matches a real project need.
2. Understand the requirements and existing architecture.
3. Implement the feature and submit a pull request.
4. Document the work so other developers can understand it.

I chose a connector issue because I wanted hands-on experience with APIs, data engineering, and AI infrastructure.

🧠 Why connectors matter for AI

Consider a company that stores information across different tools. Expense records live in a financial platform, project updates live elsewhere, and important context is buried in notes.

Even a capable AI system cannot use information it cannot access.

Cognee is an open-source project focused on turning data into structured AI memory. Data connectors help bring information from external systems into that workflow.

That led me to a practical question:

How can we make business expense information available to an AI knowledge system in a useful, structured, and maintainable way?

My contribution was an attempt to solve that problem for Brex.

🔌 What I built: the Brex connector

Brex provides business expense and budget management capabilities. My connector brings selected Brex data into Cognee's ingestion workflow.

You can explore the implementation here:

• GitHub issue #4797 — Brex connector requirements
• Pull request #389 — My implementation

The connector supports three data resources:

The interesting part is that these resources don't all need to be represented in the same way.

Expense memos can become readable documents for contextual knowledge extraction, while expenses and budgets can remain structured records for organized data processing.

This gives downstream workflows access to both business context and structured information.

⚙️ How the connector works

Here is the simplified architecture:

       Brex Platform API
                |
                v
       Authentication
       + Pagination
                |
                v
       Fetch Expenses,
       Memos and Budgets
                |
                v
       Transform the Data
          /          \
         v            v
   Document Mode   Structured Mode
         \            /
          v          v
          DLT Ingestion
                |
                v
          Cognee Workflow
                |
                v
        AI Knowledge Retrieval

Let's break down the engineering decisions behind it.

1. Authentication and API requests

The connector uses Bearer token authentication to access the Brex Platform API.

It also handles paginated responses, so the integration can retrieve records across multiple API pages rather than stopping at the first batch.

API integrations must account for failure cases, too. The implementation includes handling for authentication errors, permission errors, and rate limits.

2. Turning raw data into useful information

Raw API responses aren't always the best representation for an AI knowledge pipeline.

For example, consider an expense containing a merchant, amount, department, budget, and a note explaining why the purchase was necessary.

The connector can transform that context into a readable document card while preserving expenses and budgets as structured records.

This separation makes the data easier to consume in different downstream workflows.

3. Incremental synchronization

Fetching every record during every run can be inefficient as datasets grow.

The connector uses Brex's posted_at_start filter and stores synchronization state through DLT's resource-state mechanism to track the latest processed timestamp.

Subsequent runs can then use that timestamp to retrieve relevant new or updated expense records.

The important lesson: reliable integrations need to remember where the previous synchronization ended.

4. Handling deletions safely

One of the more interesting challenges was deletion reconciliation.

Suppose a record disappears from the source. Eventually, the destination should stop treating that record as current.

However, imagine the API temporarily fails while the connector is checking which records still exist. Treating an incomplete response as a complete snapshot could incorrectly mark valid records as deleted.

The implementation therefore includes a guardrail: if reconciliation encounters API or network errors, deletion-marker generation is aborted rather than proceeding with potentially incomplete information.

This is an important distinction between simply syncing data and building a connector that considers data integrity.

💻 What does using it look like?

The pull request includes a usage example following Cognee's API.

After installing and configuring the connector and providing the required Brex API credentials, the basic ingestion pattern looks like this:

import asyncio
import os
import cognee

from cognee_community_connector_brex import brex_source


async def main():
    source = brex_source(
        token=os.environ["BREX_API_TOKEN"],
        include=["expenses", "memos", "budgets"],
    )

    await cognee.remember(
        source,
        dataset_name="brex_spend",
        primary_key="id",
        write_disposition="merge",
        max_rows_per_table=0,
    )


if __name__ == "__main__":
    asyncio.run(main())

This example illustrates how the connector is configured and passed to Cognee's ingestion API. A complete setup also requires installing the relevant dependencies, configuring valid credentials with appropriate permissions, and checking the compatibility of the project's installed versions.

Once the data is ingested, it can be used in a Cognee search workflow according to the configured pipeline and supported query types.

For the full implementation and query example, see the pull request.

🧪 Testing is part of the feature

A connector should not be considered reliable just because it works with one successful API response.

The pull request includes offline unit tests for important paths, including:

• Authentication and API error handling.
• Pagination and data transformation.
• Incremental synchronization state.
• Deletion-marker generation and failure guardrails.

The PR also documents an end-to-end DLT pipeline run using DuckDB.

These tests matter because connectors operate at a boundary between systems. A mistake can affect not only the data being fetched but also what downstream applications believe is current.

I came away with a stronger appreciation for testing failure scenarios—not just the happy path.

📚 What I learned from this contribution

This project taught me several practical lessons.

Understand the existing architecture before coding. Following established connector patterns makes a contribution easier to review and maintain.

Think about synchronization from day one. Initial ingestion is only one part of the problem. Updates, deletion handling, and persistent state matter just as much.

Protect data when dependencies fail. A partial API response should not automatically be treated as the complete truth.

Treat documentation as part of the implementation. Clear setup instructions and realistic examples help other developers use a feature without first having to understand every line of code.

Most importantly, I learned that a meaningful open-source contribution often involves making many small engineering decisions correctly.

🙏 Thank you, Kunal Kushwaha, Cognee, and WeMakeDevs!

A big thank you to Kunal Kushwaha and the WeMakeDevs team for encouraging developers to build in public, learn by contributing, and participate in open-source initiatives like MergeTober.

Thank you to the Cognee team and community for maintaining the project and creating opportunities to work on real integration challenges.

I am also grateful to the maintainers and community members who helped with the issue assignment and contribution process.

Open source is about more than individual pull requests. It is about a community where developers can learn from one another and gradually help improve the tools everyone uses.

🔗 Explore the project

If you're interested in AI memory, data engineering, or contributing to open source, these resources are a good starting point:

• My pull request: Brex spend data-source connector
• Original issue: Add a Brex data-source connector
• Cognee: Explore the main repository
• MergeTober: Learn about the initiative

🚀 Your turn: build something useful

You don't need to start with a groundbreaking idea to contribute to open source.

Pick a real problem. Read the existing code. Ask questions when you need to. Test your implementation. Document what you learn.

That is how small contributions become useful improvements to real software.

My MergeTober journey is another step in that direction, and I'm looking forward to learning and contributing more.

Have you built an API integration or contributed to an AI open-source project? Share your experience in the comments. I'd love to learn what challenges you encountered.