How to use the Custom Tasks and Pipelines agent skill

What the skill makes the agent do

Agent skills are folders of instructions that an agent such as Claude loads when a task matches the skill's description. This one applies when you are building your own cognee processing rather than ordinary ingestion. The source says the right tool for ordinary document ingestion is remember(), so the skill is for cases cognee does not ship with, such as your own extraction, your own node types, or a post-processing step over the graph.

• Write custom tasks as plain Python functions.
• Chain tasks into a pipeline with run_custom_pipeline or the lightweight run_pipeline.
• Store custom DataPoints in the graph with add_data_points.
• Run custom enrichment over an existing graph.
• Check pipeline run status and debug how data moves between tasks.

How to install the skill

Download the skill ZIP from this page. It contains one file, SKILL.md. Then choose one of the two install routes below.

For Claude Code, unzip the folder into ~/.claude/skills/ to make it available everywhere, or into a project's .claude/skills/ folder to limit it to that project.

In the Claude apps, open Settings, go to Capabilities, then Skills, and upload the ZIP.

Prepare a working cognee environment

The skill does not replace a cognee setup. The code depends on cognee's graph and vector databases, and any LLM that a task calls needs credentials. You need Python with the cognee package installed and a configured cognee environment before the agent's code can run.

• Python with the cognee package installed.
• A configured graph database and vector database.
• LLM access, only if your tasks call an LLM.

Trigger the skill with natural requests

You do not need special syntax. Describe the processing you want and mention cognee. The skill's description matches requests about writing custom tasks, chaining pipelines, storing DataPoints, running enrichment over the graph, checking run status, or debugging data flow between tasks.

• Write a cognee task that extracts people's names from each document and store them as a Person node.
• Chain my custom task with add_data_points and run it on the people dataset.
• Enrich the existing graph of my dataset with new DataPoints using run_custom_pipeline.
• Check the status of my custom pipeline run for this dataset.
• Why does my second task receive a list instead of a single item?

Example: a custom extraction pipeline

The skill's central example defines a Person DataPoint with identity_fields so repeated runs merge instead of duplicating, writes an extraction task, and passes it with add_data_points to run_custom_pipeline. The first task receives each document as a one-element list, which the example handles with a check for string items.

• Use run_custom_pipeline for the normal case, because it runs against a dataset with permissions, a per-dataset lock, run records and status.
• Use the lightweight run_pipeline only for quick chains with no permissions, locks or run rows.

Debug how data flows between tasks

Most pipeline bugs come from a few flow rules. Each document runs the whole chain on its own. data_per_batch controls how many documents run at the same time and is not a batch size. A task's batch_size controls how the previous task's generator output is grouped before it arrives. Returning or yielding Drop removes an item from the stream, and enriches=True passes the input on unchanged when a function returns None.

• If a task gets a list when you expected one item, that is the one-element list rule.
• If nodes duplicate on every run, check that each DataPoint has identity_fields.
• If run_custom_pipeline does nothing on an older database, run await cognee.run_migrations() first, since it does not run migrations itself.
• Run status can be checked with cognee.datasets.get_status and the pipeline_names argument, because without it only cognify_pipeline is reported.