feat: add credential management

Signed-off-by: Grant Linville <[email protected]>

Author: g-linville

gptscript/credentials.py

@@ -0,0 +1,74 @@
+import json
+from datetime import datetime, timezone
+from enum import Enum
+
+
+def is_timezone_aware(dt: datetime):
+    return dt.tzinfo is not None and dt.tzinfo.utcoffset(dt) is not None
+
+
+class CredentialType(Enum):
+    Tool = "tool",
+    ModelProvider = "modelProvider"
+
+
+class Credential:
+    def __init__(self,
+                 context: str = "default",
+                 toolName: str = "",
+                 type: CredentialType = CredentialType.Tool,
+                 env: dict[str, str] = None,
+                 ephemeral: bool = False,
+                 expiresAt: datetime = None,
+                 refreshToken: str = "",
+                 ):
+        self.context = context
+        self.toolName = toolName
+        self.type = type
+        self.env = env
+        self.ephemeral = ephemeral
+        self.expiresAt = expiresAt
+        self.refreshToken = refreshToken
+
+        if self.env is None:
+            self.env = {}
+
+    def to_json(self):
+        datetime_str = ""
+
+        if self.expiresAt is not None:
+            system_tz = datetime.now().astimezone().tzinfo
+
+            if not is_timezone_aware(self.expiresAt):
+                self.expiresAt = self.expiresAt.replace(tzinfo=system_tz)
+                datetime_str = self.expiresAt.isoformat()
+
+                # For UTC only, replace the "+00:00" with "Z"
+                if self.expiresAt.tzinfo == timezone.utc:
+                    datetime_str = datetime_str.replace("+00:00", "Z")
+
+        req = {
+            "context": self.context,
+            "toolName": self.toolName,
+            "type": self.type.value[0],
+            "env": self.env,
+            "ephemeral": self.ephemeral,
+            "refreshToken": self.refreshToken,
+        }
+
+        if datetime_str != "":
+            req["expiresAt"] = datetime_str
+
+        return json.dumps(req)
+
+class CredentialRequest:
+    def __init__(self,
+                 content: str = "",
+                 allContexts: bool = False,
+                 context: str = "default",
+                 name: str = "",
+                 ):
+        self.content = content
+        self.allContexts = allContexts
+        self.context = context
+        self.name = name

gptscript/gptscript.py

@@ -9,6 +9,7 @@
 import requests
 
 from gptscript.confirm import AuthResponse
+from gptscript.credentials import Credential
 from gptscript.frame import RunFrame, CallFrame, PromptFrame, Program
 from gptscript.opts import GlobalOptions
 from gptscript.prompt import PromptResponse
@@ -183,6 +184,38 @@ async def list_models(self, providers: list[str] = None, credential_overrides: l
             {"providers": providers, "credentialOverrides": credential_overrides}
         )).split("\n")
 
+    async def list_credentials(self, context: str = "default", all_contexts: bool = False) -> list[Credential] | str:
+        res = await self._run_basic_command(
+            "credentials",
+            {"context": context, "allContexts": all_contexts}
+        )
+        if res.startswith("an error occurred:"):
+            return res
+
+        return [Credential(**c) for c in json.loads(res)]
+
+    async def create_credential(self, cred: Credential) -> str:
+        return await self._run_basic_command(
+            "credentials/create",
+            {"content": cred.to_json()}
+        )
+
+    async def reveal_credential(self, context: str = "default", name: str = "") -> Credential | str:
+        res = await self._run_basic_command(
+            "credentials/reveal",
+            {"context": context, "name": name}
+        )
+        if res.startswith("an error occurred:"):
+            return res
+
+        return Credential(**json.loads(res))
+
+    async def delete_credential(self, context: str = "default", name: str = "") -> str:
+        return await self._run_basic_command(
+            "credentials/delete",
+            {"context": context, "name": name}
+        )
+
 
 def _get_command():
     if os.getenv("GPTSCRIPT_BIN") is not None:

tests/test_gptscript.py

@@ -8,6 +8,7 @@
 import pytest
 
 from gptscript.confirm import AuthResponse
+from gptscript.credentials import Credential
 from gptscript.exec_utils import get_env
 from gptscript.frame import RunEventType, CallFrame, RunFrame, RunState, PromptFrame
 from gptscript.gptscript import GPTScript
@@ -683,3 +684,20 @@ async def test_parse_with_metadata_then_run(gptscript):
     tools = await gptscript.parse(cwd + "/tests/fixtures/parse-with-metadata.gpt")
     run = gptscript.evaluate(tools[0])
     assert "200" == await run.text(), "Expect file to have correct output"
+
+@pytest.mark.asyncio
+async def test_credentials(gptscript):
+    name = "test-" + str(os.urandom(4).hex())
+    res = await gptscript.create_credential(Credential(toolName=name, env={"TEST": "test"}))
+    assert not res.startswith("an error occurred"), "Unexpected error creating credential: " + res
+
+    res = await gptscript.list_credentials()
+    assert not str(res).startswith("an error occurred"), "Unexpected error listing credentials: " + str(res)
+    assert len(res) > 0, "Expected at least one credential"
+
+    res = await gptscript.reveal_credential(name=name)
+    assert not str(res).startswith("an error occurred"), "Unexpected error revealing credential: " + res
+    assert res.env["TEST"] == "test", "Unexpected credential value: " + str(res)
+
+    res = await gptscript.delete_credential(name=name)
+    assert not res.startswith("an error occurred"), "Unexpected error deleting credential: " + res