LangChain

pica-langchain

Install the Python SDK to unlock powerful tools for LangChain

Installation

Install the Pica LangChain SDK:

pip install pica-langchain

Configuration

The PicaClientOptions class allows you to configure the Pica client with the following options:

Option	Type	Required	Default	Description
server_url	str	No	https://api.picaos.com	URL for self-hosted Pica server.
connectors	List[str]	No	All available connectors	List of connector keys to filter by. Pass [”*”] to initialize all available connectors, or specific connector keys to filter. If empty, no connections will be initialized.
identity	str	No	None	Filter connections by specific identifier.
identity_type	`"user", "team", "project", "organization"`	No	None	Filter connections by identity type.
authkit	Boolean	No	false	If true, the SDK will use Authkit to connect to prompt the user to connect to a platform that they do not currently have access to

The create_pica_agent function allows customizing the following parameters:

Option	Type	Required	Default	Description
verbose	bool	No	False	Whether to print verbose logs.
system_prompt	str	No	None	A custom system prompt to append to the default system prompt.
agent_type	AgentType	No	OPENAI_FUNCTIONS	The type of agent to create.
tools	List[BaseTool]	No	None	A list of tools to use in the agent.
return_intermediate_steps	bool	No	False	Whether to return the intermediate steps of the agent.

Usage

This is a getting started example using Pica with LangChain and OpenAI.

Ensure you have the following environment variables set:

export PICA_SECRET="your-pica-secret"
export OPENAI_API_KEY="your-openai-api-key"

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from pica_langchain import PicaClient, create_pica_agent
from pica_langchain.models import PicaClientOptions

def main():
    try:
        pica_client = PicaClient(
            secret=os.environ["PICA_SECRET"],
            options=PicaClientOptions(
                connectors=["*"] # Initialize all available connections or pass specific connector keys
                # server_url="https://my-self-hosted-server.com",
                # identity="user-id",
                # identity_type="user"
            )
        )
        
        llm = ChatOpenAI(
            temperature=0,
            model="gpt-4o",
        )

        # Create an agent with Pica tools
        agent = create_pica_agent(
            client=pica_client,
            llm=llm,
            agent_type=AgentType.OPENAI_FUNCTIONS,
        )

        # Execute a multi-step workflow using the GitHub Connector
        result = agent.invoke({
            "input": (
                "What connections do I have access to?"
            )
        })
        
        print(f"\nWorkflow Result:\n {result}")
    
    except Exception as e:
        print(f"ERROR: An unexpected error occurred: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()

Logging

The Pica LangChain SDK uses the logging module to log messages. The log level can be set using the PICA_LOG_LEVEL environment variable.

The following log levels are available:

debug
info
warning
error
critical

export PICA_LOG_LEVEL="debug"

Examples

Code Solutions

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from pica_langchain import PicaClient, create_pica_agent
from pica_langchain.models import PicaClientOptions

pica_client = PicaClient(
    secret=os.environ["PICA_SECRET"],
    options=PicaClientOptions(
        connectors=["*"], # Initialize all available connections for this example
    )
)

llm_with_handler = ChatOpenAI(
    temperature=0,
    model="gpt-4o",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

agent_with_handler = create_pica_agent(
    client=pica_client,
    llm=llm_with_handler,
    agent_type=AgentType.OPENAI_FUNCTIONS,
)

for chunk in agent_with_handler.stream({
    "input": "List three platforms available in Pica."
}):
    print(chunk)

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from pica_langchain import PicaClient, create_pica_agent
from pica_langchain.models import PicaClientOptions

pica_client = PicaClient(
    secret=os.environ["PICA_SECRET"],
    options=PicaClientOptions(
        connectors=["*"], # Initialize all available connections for this example
    )
)

llm_with_handler = ChatOpenAI(
    temperature=0,
    model="gpt-4o",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

agent_with_handler = create_pica_agent(
    client=pica_client,
    llm=llm_with_handler,
    agent_type=AgentType.OPENAI_FUNCTIONS,
)

for chunk in agent_with_handler.stream({
    "input": "List three platforms available in Pica."
}):
    print(chunk)

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from pica_langchain import PicaClient, create_pica_agent
from pica_langchain.models import PicaClientOptions

pica_client = PicaClient(
    secret=os.environ["PICA_SECRET"],
    options=PicaClientOptions(
        authkit=True, # Enable AuthKit settings
        connectors=["*"]
    )
)

llm = ChatOpenAI(
    temperature=0,
    model="gpt-4o",
)

# Create an agent with Pica tools
agent = create_pica_agent(
    client=pica_client,
    llm=llm,
    agent_type=AgentType.OPENAI_FUNCTIONS,
    return_intermediate_steps=True # Show the intermediate steps
)

result = agent.invoke({
    "input": (
        # This will trigger the promptToConnectPlatform tool 
        # if the user doesn't have google calendar connected
        # On your client, you can listen for this tool call and prompt 
        # the user to connect via the AuthKit
        "Connect to google calendar"
    )
})

print(f"\nWorkflow Result:\n {result}")

Workflows

from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from pica_langchain import PicaClient, create_pica_agent

pica_client = PicaClient(
    secret="YOUR_PICA_SECRET",
    options=PicaClientOptions(
        connectors=["*"]
    )
)

llm = ChatOpenAI(temperature=0, model="gpt-4o")

agent = create_pica_agent(
    client=pica_client,
    llm=llm,
    agent_type=AgentType.OPENAI_FUNCTIONS
)

result = agent.invoke({
    "input": (
        "Star the picahq/pica repo in github. "
        "Then, list 5 of the repositories that I have starred in github."
    )
})

print(f"Result: {result}")

from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from pica_langchain import PicaClient, create_pica_agent

pica_client = PicaClient(
    secret="YOUR_PICA_SECRET",
    options=PicaClientOptions(
        connectors=["*"]
    )
)

llm = ChatOpenAI(temperature=0, model="gpt-4o")

agent = create_pica_agent(
    client=pica_client,
    llm=llm,
    agent_type=AgentType.OPENAI_FUNCTIONS
)

result = agent.invoke({
    "input": (
        "Star the picahq/pica repo in github. "
        "Then, list 5 of the repositories that I have starred in github."
    )
})

print(f"Result: {result}")

from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from pica_langchain import PicaClient, create_pica_agent

pica_client = PicaClient(
    secret="YOUR_PICA_SECRET",
    options=PicaClientOptions(
        connectors=["*"]
    )
)

llm = ChatOpenAI(temperature=0, model="gpt-4o")

agent = create_pica_agent(
    client=pica_client,
    llm=llm,
    agent_type=AgentType.OPENAI_FUNCTIONS
)

result = agent.invoke({
    "input": (
        "Retrieve the list of available bases from Airtable."
        "List the tasks from the Base 'My Tasks' with the status 'Todo'"
        "For each task, create a GitHub issue in the '' repository."
        "Use the task title as the issue title and the task description as the issue body."
        "Add the Airtable ticket ID for reference in the issue."
        "Assign the issue to the task's assignee if available."
        "Report the number of tasks created, skipped, or failed."
    )
})

print(f"Result: {result}")

from langchain_openai import ChatOpenAI
from langchain.agents import AgentType
from pica_langchain import PicaClient, create_pica_agent

pica_client = PicaClient(
    secret="YOUR_PICA_SECRET",
    options=PicaClientOptions(
        connectors=["*"]
    )
)

llm = ChatOpenAI(temperature=0, model="gpt-4o")

agent = create_pica_agent(
    client=pica_client,
    llm=llm,
    agent_type=AgentType.OPENAI_FUNCTIONS
)

result = agent.invoke({
    "input": (
        "List my available google spreadsheets. "
        "Retrieve the data from Spreadsheet ID '1NBp5QpEJV43Sq2P0aeo1DtYUof2BDY_YcdKcIIQTCcs'. "
        "List the content from the sheet named 'Sheet1' in the range 'A1:C5'."
        "Summarize the Comment column"
        "Send the summarized comment data to using gmail to hello@picaos.com"
    )
})

print(f"Result: {result}")

GitHub

@picahq/pica-langchain

Check out our GitHub repository to explore the code, contribute, or raise issues.

Get Started

Products

SDK

Use Cases

pica-langchain

Installation

Configuration

Usage

Logging

Examples

Code Solutions

Workflows

GitHub

@picahq/pica-langchain

Get Started

Products

SDK

Use Cases

pica-langchain

​Installation

​Configuration

​Usage

​Logging

​Examples

​Code Solutions

​Workflows

​GitHub

@picahq/pica-langchain

Installation

Configuration

Usage

Logging

Examples

Code Solutions

Workflows

GitHub