Skip to main content

Module airbyte_agent_sdk.connectors.hubspot

Hubspot connector for Airbyte SDK.

Auto-generated from OpenAPI specification.

Sub-modules​

  • airbyte_agent_sdk.connectors.hubspot.connector
  • airbyte_agent_sdk.connectors.hubspot.connector_model
  • airbyte_agent_sdk.connectors.hubspot.models
  • airbyte_agent_sdk.connectors.hubspot.types

Classes​

AirbyteAuthConfig(**data: Any) : Authentication configuration for Airbyte hosted mode execution.

Pass this to the connector's auth_config parameter to use hosted mode, where API credentials are stored securely in Airbyte Cloud.

For hosted mode execution, provide client credentials with either:

  • connector_id: Direct connector/source ID (skips lookup)
  • workspace_name: Workspace name for connector lookup

Attributes: workspace_name: Workspace name for hosted mode connector lookup organization_id: Optional Airbyte organization ID for multi-org selection airbyte_client_id: Airbyte OAuth client ID (required for hosted mode) airbyte_client_secret: Airbyte OAuth client secret (required for hosted mode) connector_id: Specific connector/source ID (skips lookup if provided)

Examples:

Hosted mode with connector_id (no lookup needed)

connector = GongConnector( auth_config=AirbyteAuthConfig( airbyte_client_id="client_abc123", airbyte_client_secret="secret_xyz789", connector_id="existing-source-uuid" ) )

Hosted mode with workspace_name (lookup by workspace)

connector = GongConnector( auth_config=AirbyteAuthConfig( workspace_name="user-123", organization_id="00000000-0000-0000-0000-000000000123", airbyte_client_id="client_abc123", airbyte_client_secret="secret_xyz789" ) )

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel

Class variables​

airbyte_client_id: str | None : The type of the None singleton.

airbyte_client_secret: str | None : The type of the None singleton.

connector_id: str | None : The type of the None singleton.

model_config : The type of the None singleton.

organization_id: str | None : The type of the None singleton.

workspace_name: str | None : The type of the None singleton.

AirbyteSearchMeta(**data: Any) : Pagination metadata for search responses.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel

Class variables​

cursor: str | None : Cursor for fetching the next page of results.

has_more: bool : Whether more results are available.

model_config : The type of the None singleton.

took_ms: int | None : Time taken to execute the search in milliseconds.

AirbyteSearchResult(**data: Any) : Result from Airbyte cache search operations with typed records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel
  • typing.Generic

Descendants​

  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult[CompaniesSearchData]
  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult[ContactsSearchData]
  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult[DealsSearchData]
  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult[TicketsSearchData]

Class variables​

data: list[~D] : List of matching records.

meta: airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchMeta : Pagination metadata.

model_config : The type of the None singleton.

CompaniesSearchResult(**data: Any) : Result from Airbyte cache search operations with typed records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult
  • pydantic.main.BaseModel
  • typing.Generic

ContactsSearchResult(**data: Any) : Result from Airbyte cache search operations with typed records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult
  • pydantic.main.BaseModel
  • typing.Generic

DealsSearchResult(**data: Any) : Result from Airbyte cache search operations with typed records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult
  • pydantic.main.BaseModel
  • typing.Generic

TicketsSearchResult(**data: Any) : Result from Airbyte cache search operations with typed records.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • airbyte_agent_sdk.connectors.hubspot.models.AirbyteSearchResult
  • pydantic.main.BaseModel
  • typing.Generic

CompaniesSearchData(**data: Any) : Search result data for companies entity.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel

Class variables​

archived: bool | None : Indicates whether the company has been deleted and moved to the recycling bin

contacts: list[typing.Any] | None : Associated contact records linked to this company

created_at: str | None : Timestamp when the company record was created

id: str | None : Unique identifier for the company record

model_config : The type of the None singleton.

properties: dict[str, typing.Any] : Object containing all property values for the company

updated_at: str | None : Timestamp when the company record was last modified

ContactsSearchData(**data: Any) : Search result data for contacts entity.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel

Class variables​

archived: bool | None : Boolean flag indicating whether the contact has been archived or deleted.

companies: list[typing.Any] | None : Associated company records linked to this contact.

created_at: str | None : Timestamp indicating when the contact was first created in the system.

id: str | None : Unique identifier for the contact record.

model_config : The type of the None singleton.

properties: dict[str, typing.Any] : Key-value object storing all contact properties and their values.

updated_at: str | None : Timestamp indicating when the contact record was last modified.

DealsSearchData(**data: Any) : Search result data for deals entity.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel

Class variables​

archived: bool | None : Indicates whether the deal has been deleted and moved to the recycling bin

companies: list[typing.Any] | None : Collection of company records associated with the deal

contacts: list[typing.Any] | None : Collection of contact records associated with the deal

created_at: str | None : Timestamp when the deal record was originally created

id: str | None : Unique identifier for the deal record

line_items: list[typing.Any] | None : Collection of product line items associated with the deal

model_config : The type of the None singleton.

properties: dict[str, typing.Any] : Key-value object containing all deal properties and custom fields

updated_at: str | None : Timestamp when the deal record was last modified

HubspotConnector(auth_config: HubspotAuthConfig | AirbyteAuthConfig | BaseModel | None = None, on_token_refresh: Any | None = None) : Type-safe Hubspot API connector.

Auto-generated from OpenAPI specification with full type safety.

Initialize a new hubspot connector instance.

Supports both local and hosted execution modes:

  • Local mode: Provide connector-specific auth config (e.g., HubspotAuthConfig)
  • Hosted mode: Provide AirbyteAuthConfig with client credentials and either connector_id or workspace_name

Args: auth_config: Either connector-specific auth config for local mode, or AirbyteAuthConfig for hosted mode on_token_refresh: Optional callback for OAuth2 token refresh persistence. Called with new_tokens dict when tokens are refreshed. Can be sync or async. Example: lambda tokens: save_to_database(tokens) Examples:

Local mode (direct API calls)

connector = HubspotConnector(auth_config=HubspotAuthConfig(client_id="...", client_secret="...", refresh_token="...", access_token="..."))

Hosted mode with explicit connector_id (no lookup needed)

connector = HubspotConnector( auth_config=AirbyteAuthConfig( airbyte_client_id="client_abc123", airbyte_client_secret="secret_xyz789", connector_id="existing-source-uuid" ) )

Hosted mode with lookup by workspace_name

connector = HubspotConnector( auth_config=AirbyteAuthConfig( workspace_name="user-123", organization_id="00000000-0000-0000-0000-000000000123", airbyte_client_id="client_abc123", airbyte_client_secret="secret_xyz789" ) )

Class variables​

connector_name : The type of the None singleton.

connector_version : The type of the None singleton.

sdk_version : The type of the None singleton.

Static methods​

configure_oauth_app_parameters(*, airbyte_config: AirbyteAuthConfig, credentials: HubspotOAuthCredentials | None) ‑> None : Configure or remove OAuth app credentials for your organization.

When credentials are provided, replaces the default Airbyte-managed OAuth app credentials with your own. After calling this, all OAuth flows for this connector in your organization will use the provided credentials.

When credentials are None, removes any existing override so the organization reverts to the default Airbyte-managed OAuth app.

Args: airbyte_config: Airbyte hosted auth config with client credentials. credentials: Your OAuth app credentials (HubspotOAuthCredentials), or None to remove the override.

Example: await HubspotConnector.configure_oauth_app_parameters( airbyte_config=AirbyteAuthConfig( airbyte_client_id="client_abc", airbyte_client_secret="secret_xyz", ), credentials=HubspotOAuthCredentials( client_id="...", client_secret="...", ), )

await HubspotConnector.configure_oauth_app_parameters( airbyte_config=AirbyteAuthConfig( airbyte_client_id="client_abc", airbyte_client_secret="secret_xyz", ), credentials=None, )

create(*, airbyte_config: AirbyteAuthConfig, auth_config: "'HubspotAuthConfig' | None" = None, server_side_oauth_secret_id: str | None = None, name: str | None = None, replication_config: dict[str, Any] | None = None, source_template_id: str | None = None) : Create a new hosted connector on Airbyte Cloud.

This factory method:

  1. Creates a source on Airbyte Cloud with the provided credentials
  2. Returns a connector configured with the new connector_id

Supports two authentication modes:

  1. Direct credentials: Provide auth_config with typed credentials
  2. Server-side OAuth: Provide server_side_oauth_secret_id from OAuth flow

Args: airbyte_config: Airbyte hosted auth config with client credentials and workspace_name. Optionally include organization_id for multi-org request routing. auth_config: Typed auth config. Required unless using server_side_oauth_secret_id. server_side_oauth_secret_id: OAuth secret ID from get_consent_url redirect. When provided, auth_config is not required. name: Optional source name (defaults to connector name + workspace_name) replication_config: Optional replication settings dict. Required for connectors with x-airbyte-replication-config (REPLICATION mode sources). source_template_id: Source template ID. Required when organization has multiple source templates for this connector type.

Returns: A HubspotConnector instance configured in hosted mode

Raises: ValueError: If neither or both auth_config and server_side_oauth_secret_id provided

Example:

Create a new hosted connector with API key auth

connector = await HubspotConnector.create( airbyte_config=AirbyteAuthConfig( workspace_name="my-workspace", organization_id="00000000-0000-0000-0000-000000000123", airbyte_client_id="client_abc", airbyte_client_secret="secret_xyz", ), auth_config=HubspotAuthConfig(client_id="...", client_secret="...", refresh_token="...", access_token="..."), )

With server-side OAuth:

connector = await HubspotConnector.create( airbyte_config=AirbyteAuthConfig( workspace_name="my-workspace", organization_id="00000000-0000-0000-0000-000000000123", airbyte_client_id="client_abc", airbyte_client_secret="secret_xyz", ), server_side_oauth_secret_id="airbyte_oauth_...secret...", )

Use the connector

result = await connector.execute("entity", "list", {})

get_consent_url(*, airbyte_config: AirbyteAuthConfig, redirect_url: str, name: str | None = None, replication_config: dict[str, Any] | None = None, source_template_id: str | None = None) ‑> str : Initiate server-side OAuth flow with auto-source creation.

Returns a consent URL where the end user should be redirected to grant access. After completing consent, the source is automatically created and the user is redirected to your redirect_url with a connector_id query parameter.

Args: airbyte_config: Airbyte hosted auth config with client credentials and workspace_name. Optionally include organization_id for multi-org request routing. redirect_url: URL where users will be redirected after OAuth consent. After consent, user arrives at: redirect_url?connector_id=... name: Optional name for the source. Defaults to connector name + workspace_name. replication_config: Optional replication settings dict. Merged with OAuth credentials. source_template_id: Source template ID. Required when organization has multiple source templates for this connector type.

Returns: The OAuth consent URL

Example: consent_url = await HubspotConnector.get_consent_url( airbyte_config=AirbyteAuthConfig( workspace_name="my-workspace", organization_id="00000000-0000-0000-0000-000000000123", airbyte_client_id="client_abc", airbyte_client_secret="secret_xyz", ), redirect_url="https://myapp.com/oauth/callback", name="My Hubspot Source", )

Redirect user to: consent_url

After consent, user arrives at: https://myapp.com/oauth/callback?connector_id=...

tool_utils(func: _F | None = None, *, update_docstring: bool = True, max_output_chars: int | None = 100000, framework: FrameworkName | None = None, internal_retries: int = 0, should_internal_retry: Callable[[Exception, tuple[Any, ...], dict[str, Any]], bool] | None = None, exhausted_runtime_failure_message: Callable[[Exception, tuple[Any, ...], dict[str, Any]], str | None] | None = None) ‑> ~_F | Callable[[~_F], ~_F] : Decorator that adds tool utilities like docstring augmentation and output limits.

Composes :func:airbyte_agent_sdk.translation.translate_exceptions for runtime wrapping (sync/async branch + output-size check + framework signal translation + optional internal retry loop), and adds connector-specific docstring augmentation on top of it.

Usage: @mcp.tool() @HubspotConnector.tool_utils async def execute(entity: str, action: str, params: dict): ...

@mcp.tool() @HubspotConnector.tool_utils(update_docstring=False, max_output_chars=None) async def execute(entity: str, action: str, params: dict): ...

@mcp.tool() @HubspotConnector.tool_utils(framework="pydantic_ai", internal_retries=2) async def execute(entity: str, action: str, params: dict): ...

Args: update_docstring: When True, append connector capabilities to doc. max_output_chars: Max serialized output size before raising. Use None to disable. framework: One of "pydantic_ai" | "langchain" | "openai_agents" | "mcp". Defaults to None → auto-detect by attempting each framework's canonical import in order. Explicit always wins. internal_retries: How many transient runtime failures (429/5xx, network, timeout) to retry silently before surfacing. Default 0. Forwarded to :func:airbyte_agent_sdk.translation.translate_exceptions. should_internal_retry: Optional predicate (error, args, kwargs) -> bool further restricting which retryable errors are safe for this specific tool. Forwarded to :func:airbyte_agent_sdk.translation.translate_exceptions. exhausted_runtime_failure_message: Optional callback (error, args, kwargs) -> str | None. Invoked after internal retries are exhausted OR were skipped via should_internal_retry returning False. Forwarded to :func:airbyte_agent_sdk.translation.translate_exceptions.

Instance variables​

connector_id: str | None : Get the connector/source ID (only available in hosted mode).

Returns: The connector ID if in hosted mode, None if in local mode.

Example: connector = await HubspotConnector.create(...) print(f"Created connector: {connector.connector_id}")

Methods​

check(self) ‑> airbyte_agent_sdk.connectors.hubspot.models.HubspotCheckResult : Perform a health check to verify connectivity and credentials.

Executes a lightweight list operation (limit=1) to validate that the connector can communicate with the API and credentials are valid.

Returns: HubspotCheckResult with status ("healthy" or "unhealthy") and optional error message

Example: result = await connector.check() if result.status == "healthy": print("Connection verified!") else: print(f"Check failed: {result.error}")

close(self) : Close the connector and release resources.

entity_schema(self, entity: str) ‑> dict[str, typing.Any] | None : Get the JSON schema for an entity.

Args: entity: Entity name (e.g., "contacts", "companies")

Returns: JSON schema dict describing the entity structure, or None if not found.

Example: schema = connector.entity_schema("contacts") if schema: print(f"Contact properties: {list(schema.get('properties', {}).keys())}")

execute(self, entity: str, action: "Literal['list', 'get', 'api_search', 'context_store_search']", params: Mapping[str, Any] | None = None) ‑> Any : Execute an entity operation with full type safety.

This is the recommended interface for blessed connectors as it:

  • Uses the same signature as non-blessed connectors
  • Provides full IDE autocomplete for entity/action/params
  • Makes migration from generic to blessed connectors seamless

Args: entity: Entity name (e.g., "customers") action: Operation action (e.g., "create", "get", "list") params: Operation parameters (typed based on entity+action)

Returns: Typed response based on the operation

Example: customer = await connector.execute( entity="customers", action="get", params={"id": "cus_123"} )

list_entities(self) ‑> list[dict[str, typing.Any]] : Get structured data about available entities, actions, and parameters.

Returns a list of entity descriptions with:

  • entity_name: Name of the entity (e.g., "contacts", "deals")
  • description: Entity description from the first endpoint
  • available_actions: List of actions (e.g., ["list", "get", "create"])
  • parameters: Dict mapping action -> list of parameter dicts

Example: entities = connector.list_entities() for entity in entities: print(f"{entity['entity_name']}: {entity['available_actions']}")

HubspotOAuthCredentials(**data: Any) : HubSpot OAuth App Credentials - Provide your own HubSpot OAuth app credentials to override the default Airbyte-managed ones.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel

Class variables​

client_id: str : Your HubSpot OAuth app's client ID

client_secret: str : Your HubSpot OAuth app's client secret

model_config : The type of the None singleton.

TicketsSearchData(**data: Any) : Search result data for tickets entity.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors (in MRO)​

  • pydantic.main.BaseModel

Class variables​

archived: bool | None : Indicates whether the ticket has been deleted and moved to the recycling bin

companies: list[typing.Any] | None : Collection of company records associated with the ticket

contacts: list[typing.Any] | None : Collection of contact records associated with the ticket

created_at: str | None : Timestamp when the ticket record was originally created

id: str | None : Unique identifier for the ticket record

model_config : The type of the None singleton.

properties: dict[str, typing.Any] : Key-value object containing all ticket properties and custom fields

updated_at: str | None : Timestamp when the ticket record was last modified

Was this page helpful?