Notion connector — syncs pages via the Notion REST API.
Uses a Notion internal integration token stored locally. All network calls
are isolated in module-level functions (_notion_api_*) to make them
trivially mockable in tests.
Users create an internal integration at https://www.notion.so/my-integrations
and paste the generated token to authenticate.
Classes
NotionConnector(token: str = '', credentials_path: str = '')
Bases: BaseConnector
Connector that syncs pages from Notion via the REST API.
Authentication uses a Notion internal integration token. Users create
an integration at https://www.notion.so/my-integrations and paste the
generated secret token.
| PARAMETER |
DESCRIPTION |
token
|
Notion integration token. If provided, it takes priority over any
stored credentials file.
TYPE:
str
DEFAULT:
''
|
credentials_path
|
Path to the JSON file where the token is stored. Defaults to
~/.openjarvis/connectors/notion.json.
TYPE:
str
DEFAULT:
''
|
Source code in src/openjarvis/connectors/notion.py
| def __init__(
self,
token: str = "",
credentials_path: str = "",
) -> None:
self._token: str = token
self._credentials_path: str = credentials_path or _DEFAULT_CREDENTIALS_PATH
self._items_synced: int = 0
self._items_total: int = 0
self._last_sync: Optional[datetime] = None
self._last_cursor: Optional[str] = None
|
Functions
Return True if a valid token is available.
Source code in src/openjarvis/connectors/notion.py
| def is_connected(self) -> bool:
"""Return ``True`` if a valid token is available."""
return bool(self._resolve_token())
|
Clear the in-memory token and delete the stored credentials file.
Source code in src/openjarvis/connectors/notion.py
| def disconnect(self) -> None:
"""Clear the in-memory token and delete the stored credentials file."""
self._token = ""
delete_tokens(self._credentials_path)
|
Return the URL where users can create a Notion integration token.
Source code in src/openjarvis/connectors/notion.py
| def auth_url(self) -> str:
"""Return the URL where users can create a Notion integration token."""
return "https://www.notion.so/my-integrations"
|
handle_callback(code: str) -> None
Persist the integration token to the credentials file.
The code parameter holds the raw token string (pasted by the user).
Source code in src/openjarvis/connectors/notion.py
| def handle_callback(self, code: str) -> None:
"""Persist the integration token to the credentials file.
The *code* parameter holds the raw token string (pasted by the user).
"""
save_tokens(self._credentials_path, {"token": code})
|
sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]
Yield :class:Document objects for Notion pages.
Paginates through /v1/search and fetches block content for each
page, rendering it to markdown.
| PARAMETER |
DESCRIPTION |
since
|
If provided, skip pages whose last_edited_time is before this
datetime.
TYPE:
Optional[datetime]
DEFAULT:
None
|
cursor
|
next_cursor from a previous sync to resume pagination.
TYPE:
Optional[str]
DEFAULT:
None
|
Source code in src/openjarvis/connectors/notion.py
| def sync(
self,
*,
since: Optional[datetime] = None,
cursor: Optional[str] = None,
) -> Iterator[Document]:
"""Yield :class:`Document` objects for Notion pages.
Paginates through ``/v1/search`` and fetches block content for each
page, rendering it to markdown.
Parameters
----------
since:
If provided, skip pages whose ``last_edited_time`` is before this
datetime.
cursor:
``next_cursor`` from a previous sync to resume pagination.
"""
token = self._resolve_token()
if not token:
return
page_cursor: Optional[str] = cursor
synced = 0
while True:
search_resp = _notion_api_search(token, cursor=page_cursor)
pages: List[Dict[str, Any]] = search_resp.get("results", [])
for page in pages:
page_id: str = page.get("id", "")
if not page_id:
continue
last_edited_str: str = page.get("last_edited_time", "")
timestamp = _parse_iso_datetime(last_edited_str)
# Apply since filter
if since is not None:
# Make since timezone-aware for comparison if needed
since_aware = since
if since.tzinfo is None and timestamp.tzinfo is not None:
since_aware = since.replace(tzinfo=timezone.utc)
if timestamp < since_aware:
continue
title = _extract_page_title(page)
url: Optional[str] = page.get("url")
blocks = _notion_api_get_blocks(token, page_id)
content = _render_blocks_to_markdown(blocks)
doc = Document(
doc_id=f"notion:{page_id}",
source="notion",
doc_type="document",
content=content,
title=title,
timestamp=timestamp,
url=url,
metadata={"page_id": page_id},
)
synced += 1
yield doc
has_more: bool = search_resp.get("has_more", False)
if not has_more:
self._last_cursor = None
break
page_cursor = search_resp.get("next_cursor")
self._last_cursor = page_cursor
self._items_synced = synced
self._last_sync = datetime.now(tz=timezone.utc)
|
Return sync progress from the most recent :meth:sync call.
Source code in src/openjarvis/connectors/notion.py
| def sync_status(self) -> SyncStatus:
"""Return sync progress from the most recent :meth:`sync` call."""
return SyncStatus(
state="idle",
items_synced=self._items_synced,
last_sync=self._last_sync,
cursor=self._last_cursor,
)
|
Expose two MCP tool specs for real-time Notion queries.
Source code in src/openjarvis/connectors/notion.py
| def mcp_tools(self) -> List[ToolSpec]:
"""Expose two MCP tool specs for real-time Notion queries."""
return [
ToolSpec(
name="notion_search_pages",
description=(
"Search Notion pages accessible to the integration by keyword. "
"Returns matching page titles and URLs."
),
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query string",
},
"max_results": {
"type": "integer",
"description": "Maximum number of pages to return",
"default": 20,
},
},
"required": ["query"],
},
category="knowledge",
),
ToolSpec(
name="notion_get_page",
description=(
"Retrieve the full markdown content of a Notion page by its ID."
),
parameters={
"type": "object",
"properties": {
"page_id": {
"type": "string",
"description": "Notion page UUID",
},
},
"required": ["page_id"],
},
category="knowledge",
),
]
|
Functions