Asana

This connector is optimized for AI agents. For the data replication connector, see Asana.

The Asana agent connector is a Python package that equips AI agents to interact with Asana through strongly typed, well-documented tools. It's ready to use directly in your Python app, in an agent framework, or exposed through an MCP.

Asana is a work management platform that helps teams organize, track, and manage projects and tasks. This connector provides access to tasks, projects, workspaces, teams, and users for project tracking, workload analysis, and productivity insights.

Example prompts

The Asana connector is optimized to handle prompts like these.

What tasks are assigned to me this week?
List all projects in my workspace
Show me the tasks for a recent project
Who are the team members in one of my teams?
Show me details of my current workspace and its users
Create a new task called 'Review Q3 report' in my project
Mark the task 'Submit proposal' as completed
Update the due date of task X to next Friday
Create a new project called 'Product Launch' in my workspace
Add a comment on the task saying 'Looks good, approved!'
Assign the task to me and set the due date to tomorrow
Delete the project 'Old Campaign'
Schedule a new team meeting as a task for next Tuesday
Add a new team member to my workspace by email
Delete the task 'Outdated draft'
Summarize my team's workload and task completion rates
Find all tasks related to {client_name} across my workspaces
Analyze the most active projects in my workspace last month
Compare task completion rates between my different teams
Identify overdue tasks across all my projects
Create a new section called 'In Review' in my project
Move a task to the 'Done' section
List all tasks in the 'To do' section
Rename the 'Backlog' section to 'Icebox'
Delete the empty 'Old Section' from the project
Create a tag called 'Urgent' in my workspace
Tag this task with 'Bug'
Remove the 'Low Priority' tag from this task
List all tasks tagged 'Release v2'
Rename the tag 'WIP' to 'In Progress'
Delete the tag 'Deprecated'

Unsupported prompts

The Asana connector isn't currently able to handle prompts like these.

Move this task to another project

Entities and actions

This connector supports the following entities and actions. For more details, see this connector's full reference documentation.

Entity	Actions
Tasks	List, Create, Get, Update, Delete, Context Store Search
Project Tasks	List
Workspace Task Search	List
Projects	List, Create, Get, Update, Delete, Context Store Search
Task Projects	List
Team Projects	List
Workspace Projects	List
Workspaces	List, Get, Context Store Search
Users	List, Get, Context Store Search
Workspace Users	List
Team Users	List
Teams	Get, Context Store Search
Workspace Teams	List
User Teams	List
Attachments	List, Get, Download, Context Store Search
Workspace Tags	List, Create
Tags	Get, Update, Delete, Context Store Search
Tag Tasks	List
Project Sections	List, Create
Sections	Get, Update, Delete, Context Store Search
Section Tasks	List, Create
Task Subtasks	List
Task Dependencies	List
Task Dependents	List
Task Stories	Create
Task Tags	Create, Delete
Workspace Memberships	Create

Asana API docs

See the official Asana API reference.

SDK installation

uv pip install airbyte-agent-sdk

SDK usage

Connectors can run in hosted or open source mode.

Hosted

In hosted mode, API credentials are stored securely in Airbyte Agents. You provide your Airbyte credentials instead. If your Airbyte client can access multiple organizations, also set organization_id.

This example assumes you've already authenticated your connector with Airbyte. See Authentication to learn more about authenticating. If you need a step-by-step guide, see the hosted execution tutorial.

The connect() factory returns a fully typed AsanaConnector and reads AIRBYTE_CLIENT_ID / AIRBYTE_CLIENT_SECRET from the environment:

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.asana import AsanaConnector

connector = connect("asana", workspace_name="<your_workspace_name>")

agent = Agent("openai:gpt-4o")

@agent.tool_plain
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.asana import AsanaConnector

connector = connect("asana", workspace_name="<your_workspace_name>")

@tool
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.asana import AsanaConnector

connector = connect("asana", workspace_name="<your_workspace_name>")

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@AsanaConnector.tool_utils(framework="openai_agents")
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Asana Assistant", tools=[asana_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk import connect
from airbyte_agent_sdk.connectors.asana import AsanaConnector

connector = connect("asana", workspace_name="<your_workspace_name>")

mcp = FastMCP("Asana Agent")

@mcp.tool
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Or pass credentials explicitly (equivalent, useful when you're not loading them from the environment):

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = AsanaConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

agent = Agent("openai:gpt-4o")

@agent.tool_plain
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = AsanaConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

@tool
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = AsanaConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@AsanaConnector.tool_utils(framework="openai_agents")
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Asana Assistant", tools=[asana_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.types import AirbyteAuthConfig

connector = AsanaConnector(
    auth_config=AirbyteAuthConfig(
        workspace_name="<your_workspace_name>",
        organization_id="<your_organization_id>",  # Optional for multi-org clients
        airbyte_client_id="<your-client-id>",
        airbyte_client_secret="<your-client-secret>"
    )
)

mcp = FastMCP("Asana Agent")

@mcp.tool
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Open source

In open source mode, you provide API credentials directly to the connector.

Pydantic AI
LangChain
OpenAI Agents
FastMCP

Pydantic AI
from pydantic_ai import Agent
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.connectors.asana.models import AsanaPersonalAccessTokenAuthConfig

connector = AsanaConnector(
    auth_config=AsanaPersonalAccessTokenAuthConfig(
        token="<Your Asana Personal Access Token. Generate one at https://app.asana.com/0/my-apps>"
    )
)

agent = Agent("openai:gpt-4o")

@agent.tool_plain
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    return await connector.execute(entity, action, params or {})

LangChain
from langchain_core.tools import tool
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.connectors.asana.models import AsanaPersonalAccessTokenAuthConfig

connector = AsanaConnector(
    auth_config=AsanaPersonalAccessTokenAuthConfig(
        token="<Your Asana Personal Access Token. Generate one at https://app.asana.com/0/my-apps>"
    )
)

@tool
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    # connector.execute returns a Pydantic envelope for typed actions; fall back to raw data otherwise.
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

OpenAI Agents
from agents import Agent, function_tool
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.connectors.asana.models import AsanaPersonalAccessTokenAuthConfig

connector = AsanaConnector(
    auth_config=AsanaPersonalAccessTokenAuthConfig(
        token="<Your Asana Personal Access Token. Generate one at https://app.asana.com/0/my-apps>"
    )
)

# strict_mode=False because `params: dict` is permissive and the default strict
# JSON schema rejects objects with additionalProperties.
@function_tool(strict_mode=False)
@AsanaConnector.tool_utils(framework="openai_agents")
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

agent = Agent(name="Asana Assistant", tools=[asana_execute])

FastMCP
from fastmcp import FastMCP
from airbyte_agent_sdk.connectors.asana import AsanaConnector
from airbyte_agent_sdk.connectors.asana.models import AsanaPersonalAccessTokenAuthConfig

connector = AsanaConnector(
    auth_config=AsanaPersonalAccessTokenAuthConfig(
        token="<Your Asana Personal Access Token. Generate one at https://app.asana.com/0/my-apps>"
    )
)

mcp = FastMCP("Asana Agent")

@mcp.tool
@AsanaConnector.tool_utils
async def asana_execute(entity: str, action: str, params: dict | None = None):
    """Execute Asana connector operations."""
    result = await connector.execute(entity, action, params or {})
    return result.model_dump(mode="json") if hasattr(result, "model_dump") else result

Authentication

For all authentication options, see the connector's authentication documentation.

Version information

Connector version: 0.1.20

Example prompts​

Unsupported prompts​

Entities and actions​

Asana API docs​

SDK installation​

SDK usage​

Hosted​

Open source​

Authentication​

Version information​