OpenAI Function Calling In Python

Exploring the AI Automation Engineer Toolkit with the Jira SDK, Vaul and Langfuse.

Function Calls as Custom Actions in a GPT

Want to keep up with the latest AI research and need a more streamlined approach? Textlayer AI is the first purpose-built research platform for developers the gives you free access to personalized recommendations, easy-to-read summaries, and full chat with implementation support.

The ability to use Function Calling/Tools is probably my favourite advance in LLM development over the past few years.

Allowing a model to reason whether or not it should execute a function unlocks some incredibly powerful workflows that would have taken an entire ML team before.

But even with that, working with Tools — understanding how they’re used, how to write JSON schema, and then trying to implement them — can be pretty overwhelming as you’re first getting started.

Luckily there are a bunch of tools you can use that drastically simplify the process, and I thought I’d make a quick tutorial so you can implement these tools and techniques yourself.

Function Calls as Custom Actions in a GPT

What We’ll Cover

In this tutorial, we’re going to be building a basic toolkit for you to interact with Jira in managing tickets. Once you’re done, you’ll be able to use this code in your own system as part of a pipeline — maybe taking action items from a meeting and updating the status of tickets or assigning them out of a huddle.

For me, I’ve adapted the code to make a GPT that I can drag-and-drop meeting notes and convert them into tickets automatically as you see in the video above.

To do so, we’ll be implementing the Jira SDK as a set of discrete functions, using Vaul to automatically generate the JSON schema needed to convert the functions to tools, and implementing monitoring/tracing using Langfuse to make them production ready.

Tools and Tech

• Jira SDK
• Vaul
• Langfuse

Building the Functions

Let’s start off by first installing the Jira Python SDK and python-dotenv to manage our environment variables:

pip install jira python-dotenv

After that, we’ll create a Python file named jira_functions.py and add our functions for interaction with Jira. This will be CRUD (Create/Read/Update Delete) for Issues/Tickets, as well as functions for getting the comments, and dealing with transitions such as moving from “Todo” to “In Progress”.

from jira import JIRA

from dotenv import load_dotenv

import os

load_dotenv('.env')

jira = JIRA(
    server=os.environ.get("JIRA_URL"),
    basic_auth=(
        os.environ.get("JIRA_USER"),
        os.environ.get("JIRA_API_TOKEN")
    )
)

def create_issue(summary: str, description: str, issue_type: str) -> dict:
    """
    Creates a Jira issue.
    :param summary: The issue summary
    :param description: The issue description
    :param issue_type: The issue type
    :return: The created issue
    """
    try:
        new_issue = jira.create_issue(
            fields={
                "project": {"key": os.environ.get('JIRA_PROJECT_KEY')},
                "summary": summary,
                "description": description,
                "issuetype": {"name": issue_type}
            }
        )
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "id": new_issue.id,
        "key": new_issue.key,
        "summary": new_issue.fields.summary,
        "description": new_issue.fields.description,
        "type": new_issue.fields.issuetype.name
    }

def update_issue(issue_id: str, summary: str, description: str, issue_type: str) -> dict:
    """
    Updates a Jira issue.
    :param issue_id: The issue ID
    :param summary: The issue summary
    :param description: The issue description
    :param issue_type: The issue type
    :return: The updated issue
    """
    try:
        issue = jira.issue(issue_id)

        fields = {
            "summary": summary if summary else issue.fields.summary,
            "description": description if description else issue.fields.description,
            "issuetype": {"name": issue_type if issue_type else issue.fields.issuetype.name}
        }

        issue.update(fields=fields)
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "id": issue.id,
        "key": issue.key,
        "summary": issue.fields.summary,
        "description": issue.fields.description,
        "type": issue.fields.issuetype.name
    }

def delete_issue(issue_id: str) -> dict:
    """
    Deletes a Jira issue.
    :param issue_id: The issue ID
    """
    try:
        jira.issue(issue_id).delete()
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "message": "Issue deleted successfully"
    }

def get_issue(issue_id: str) -> dict:
    """
    Gets a Jira issue.
    :param issue_id: The issue ID
    :return: The issue
    """
    try:
        issue = jira.issue(issue_id)
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "id": issue.id,
        "key": issue.key,
        "summary": issue.fields.summary,
        "description": issue.fields.description,
        "type": issue.fields.issuetype.name
    }

def get_issues(project: str) -> dict:
    """
    Gets all issues in a project.
    :param project: The project key
    :return: The issues
    """
    try:
        issues = jira.search_issues(f"project={project}")
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "issues": [
            {
                "id": issue.id,
                "key": issue.key,
                "summary": issue.fields.summary,
                "description": issue.fields.description,
                "type": issue.fields.issuetype.name
            } for issue in issues
        ]
    }

def get_issue_comments(issue_id: str) -> dict:
    """
    Gets the comments for a Jira issue.
    :param issue_id: The issue ID
    :return: The comments
    """
    try:
        comments = jira.comments(issue_id)
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        'comments': [
            {
                'id': comment.id,
                'body': comment.body
            } for comment in comments
        ]
    }

def get_issue_transitions(issue_id: str) -> dict:
    """
    Gets the transitions for a Jira issue.
    :param issue_id: The issue ID
    :return: The transitions
    """
    try:
        transitions = jira.transitions(issue_id)
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        'transitions': [
            {
                'id': transition['id'],
                'name': transition['name']
            } for transition in transitions
        ]
    }

def transition_issue(issue_id: str, transition_id: str) -> dict:
    """
    Transitions a Jira issue.
    :param issue_id: The issue ID
    :param transition_id: The transition ID
    :return: The transition result
    """
    try:
        jira.transition_issue(issue_id, transition_id)
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "message": "Issue transitioned successfully"
    }

Now we’ll need to setup our environment variables as an .env file in the same directory as jira_functions.py

JIRA_URL=https://your-jira-instance.atlassian.net
JIRA_USER=your-jira-username
JIRA_API_TOKEN=your-jira-api-token
JIRA_PROJECT_KEY=your-jira-project-key

Make sure to replace the placeholders with your actual Jira instance URL, username, API key and Project Key.

Next up, let’s take a second to test the functions by creating a new issue in a test.py file to make sure everything is working properly:

from jira_functions import create_issue

issue = create_issue(
    summary="Test Issue",
    description="This is a test issue",
    issue_type="Task"
)

print(issue)

After running the script, you should see the newly created issue details printed to the console. Once you’ve verified everything is working well, feel free to delete the test.py file.

Using Vaul To Create Tools

With our core functions ready, let’s make them into tools using the Vaul tool_call decorator. This library uses a decorator that takes the type hints, function name and arguments to generate the JSON schema needed by the OpenAI API.

First up, install the Vaul package:

pip install vaul

Then, just add the @tool_call decorator to each function you want to make executable. Here’s an example:

from vaul import tool_call

@tool_call
def create_issue(summary: str, description: str, issue_type: str) -> dict:
    """
    Creates a Jira issue.
    :param summary: The issue summary
    :param description: The issue description
    :param issue_type: The issue type
    :return: The created issue
    """
    try:
        new_issue = jira.create_issue(
            fields={
                "project": {"key": os.environ.get('JIRA_PROJECT_KEY')},
                "summary": summary,
                "description": description,
                "issuetype": {"name": issue_type}
            }
        )
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "id": new_issue.id,
        "key": new_issue.key,
        "summary": new_issue.fields.summary,
        "description": new_issue.fields.description,
        "type": new_issue.fields.issuetype.name
    }

Once you have added the decorator to each of the function, we can start setting up our OpenAI session and toolkit.

from openai import OpenAI

openai_session = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY")
)

# ... existing code ...

# Create a toolkit
toolkit = {
    "create_issue": create_issue,
    "update_issue": update_issue,
    "delete_issue": delete_issue,
    "get_issue": get_issue,
    "get_issues": get_issues,
    "get_issue_comments": get_issue_comments,
    "get_issue_transitions": get_issue_transitions,
    "transition_issue": transition_issue
}

# Send a message to the OpenAI API to get all the issues
response = openai_session.chat.completions.create(
    model="gpt-3.5-turbo",
    messages=[
        {
            "role": "system",
            "content": "You are a Jira bot that can create, update, and delete issues. You can also get issue details, transitions, and comments."
        },
        {
            "role": "user",
            "content": f"Get all the issues in the {os.environ.get('JIRA_PROJECT_KEY')} project"
        }
    ],
    tools=[{
        "type": "function",
        "function": value.tool_call_schema
    } for key, value in toolkit.items()],
)

# Identify the tool call, if any
try:
    tool_call = response.choices[0].message.tool_calls[0].function.name
except AttributeError:
    tool_call = None

# Run the tool if it exists
if tool_call and tool_call in toolkit:
    tool_run = toolkit[tool_call].from_response(response, throw_error=False)
    print(tool_run)

While there isn’t a standard way to create a toolkit, my preference is using a dictionary as it is the most straightforward and readable. You can then plug it into the tools section and iterate through:

tools=[{
        "type": "function",
        "function": value.tool_call_schema
    } for key, value in toolkit.items()]

Finally, we take the response from the chat completion, check to see if a function call has been made, and run the tool, if it exists in our toolkit:

# Identify the tool call, if any
try:
    tool_call = response.choices[0].message.tool_calls[0].function.name
except AttributeError:
    tool_call = None

# Run the tool if it exists
if tool_call and tool_call in toolkit:
    tool_run = toolkit[tool_call].from_response(response, throw_error=False)
    print(tool_run)

Monitoring with Langfuse

So far in our code we have been simply printing our outputs to the console. Once we deploy however, this isn’t going to cut it.

This is where Langfuse comes in.

Langfuse is a dead-simple tool that I love working with because it doesn’t rely on a ton of overhead. A basic one-to-one swap out for the OpenAI session with their wrapper, a decorator for our functions, and we have production grade logging.

You can find out more about them and sign up here, and I’ve found their documentation to be impressively complete.

To get started, we’ll install the Langfuse package:

pip install langfuse

Logging LLM Executions

Next up, we’ll swap out the OpenAI session with the wrapper from Langfuse:

# Remove the "from openai import OpenAI" line, and replace it with the following
from langfuse.openai import OpenAI

And update our environment variables with those required by Langfuse:

LANGFUSE_PUBLIC_KEY=your-langfuse-public-key
LANGFUSE_SECRET_KEY=your-langfuse-secret-key
LANGFUSE_HOST=your-langfuse-host

… and that’s it.

Now whenever your OpenAI session runs, the entire LLM request is logged into the dashboard for you to go and inspect. They also have a playground for you to go and mess around with to help in optimizing your prompts.

Adding Monitoring to the Functions

Our last step is to add some logging/monitoring directly to the functions themselves. This will help us in viewing the direct input and output of any tool calls to debug issues.

To do so we just need to add an @observe decorator, being sure to place it underneath the @tool_call decorator.

from langfuse.decorators import observe

# Make sure to put the @observe decorator below the @tool_call decorator
@tool_call
@observe
def create_issue(summary: str, description: str, issue_type: str) -> dict:
    """
    Creates a Jira issue.
    :param summary: The issue summary
    :param description: The issue description
    :param issue_type: The issue type
    :return: The created issue
    """
    try:
        new_issue = jira.create_issue(
            fields={
                "project": {"key": os.environ.get('JIRA_PROJECT_KEY')},
                "summary": summary,
                "description": description,
                "issuetype": {"name": issue_type}
            }
        )
    except Exception as e:
        return {
            "error": str(e)
        }

    return {
        "id": new_issue.id,
        "key": new_issue.key,
        "summary": new_issue.fields.summary,
        "description": new_issue.fields.description,
        "type": new_issue.fields.issuetype.name
    }

This provides all the args, kwargs and returned output to be able to ensure your functions are running smoothly.

Conclusion

With that, you now have a fully functional script for interacting with Jira to manage your tickets. Hopefully you found this tutorial helpful — please let me know if there’s any other tools you’d like to see or questions about how to work with them, and I’d be happy to try and help.

To see the entirety of the code as a gist here, and please be sure to add some applause and share.

Happy Hacking!

Thank you for reading, and if you’d like to keep up on all the newest Data Science and ML papers, be sure to get your free account at Textlayer AI