wehub-resource-sync
f2306a3d11
Lint and Test / lint_test (3.11) (push) Has been cancelled
Lint and Test / lint_test (3.10) (push) Has been cancelled
Lint and Test / lint_test (3.12) (push) Has been cancelled
Publish to PyPI and release / Build distribution (push) Has been cancelled
Codespell / Check for spelling errors (push) Has been cancelled
Publish to PyPI and release / Publish to PyPI (push) Has been cancelled
Publish to PyPI and release / Make release (push) Has been cancelled
43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
import subprocess
|
|
from typing import Any, Dict
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class Function(BaseModel):
|
|
"""
|
|
Executes a shell command and returns the output (result).
|
|
"""
|
|
|
|
shell_command: str = Field(
|
|
...,
|
|
example="ls -la",
|
|
description="Shell command to execute.",
|
|
) # type: ignore
|
|
|
|
@classmethod
|
|
def execute(cls, shell_command: str) -> str:
|
|
process = subprocess.Popen(
|
|
shell_command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
|
|
)
|
|
output, _ = process.communicate()
|
|
exit_code = process.returncode
|
|
return f"Exit code: {exit_code}, Output:\n{output.decode()}"
|
|
|
|
@classmethod
|
|
def openai_schema(cls) -> Dict[str, Any]:
|
|
"""Generate OpenAI function schema from Pydantic model."""
|
|
schema = cls.model_json_schema()
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": "execute_shell_command",
|
|
"description": cls.__doc__.strip() if cls.__doc__ else "",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": schema.get("properties", {}),
|
|
"required": schema.get("required", []),
|
|
},
|
|
},
|
|
}
|