API Reference
This page contains the API reference for the pctx-client package.
Pctx Client
- class pctx_client.Pctx(tools: list[Tool | AsyncTool] | None = None, servers: list[HttpServerConfig | StdioServerConfig] | None = None, url: str = 'http://localhost:8080', api_key: str | None = None, execute_timeout: float = 30.0)[source]
Bases:
objectPCTX Client
Execute TypeScript/JavaScript code with access to both MCP tools and local Python tools.
- __init__(tools: list[Tool | AsyncTool] | None = None, servers: list[HttpServerConfig | StdioServerConfig] | None = None, url: str = 'http://localhost:8080', api_key: str | None = None, execute_timeout: float = 30.0)[source]
Initialize the PCTX client.
- Parameters:
tools – List of local Python tools to register
servers – List of MCP servers to register. Each server can be either: - HTTP server: {“name”: “…”, “url”: “…”, “auth”: {…}} - stdio server: {“name”: “…”, “command”: “…”, “args”: […], “env”: {…}}
url – PCTX server URL (default: http://localhost:8080)
execute_timeout – Timeout for code execution in seconds (default: 30.0)
- async list_functions() ListFunctionsOutput[source]
List all available functions organized by namespace.
This is typically the first method you should call to discover what functions are available in the current session, including both registered local tools and MCP server functions.
- Returns:
- An object containing function signatures organized
by namespace. The code attribute contains TypeScript code with function declarations that can be used for reference.
- Return type:
ListFunctionsOutput
- Raises:
SessionError – If called before establishing a session via connect().
Example
>>> async with Pctx() as pctx: ... functions = await pctx.list_functions() ... print(functions.code) # TypeScript declarations
- async search_functions(query: str, k: int = 10) list[ListedFunction][source]
Search available functions matching query.
This is typically the first method you should call to discover what functions are available in the current session, including both registered local tools and MCP server functions.
- Parameters:
query – Search query string to find relevant functions.
k – Max number of top results to return (default: 5).
- Returns:
An list of matching function signatures matching the query
- Return type:
list[ListedFunction]
- Raises:
ImportError – If bm25s is not installed.
SessionError – If called before establishing a session via connect().
- async get_function_details(functions: list[str]) GetFunctionDetailsOutput[source]
Get detailed information about specific functions.
After discovering available functions with list_functions(), use this method to get comprehensive details about parameter types, return values, and usage for the specific functions you need.
- Parameters:
functions – List of function names in ‘namespace.functionName’ format (e.g., [‘Notion.apiPostSearch’, ‘Weather.getCurrentWeather’]).
- Returns:
- An object containing detailed TypeScript
declarations for the requested functions. The code attribute contains the full function signatures with JSDoc comments.
- Return type:
GetFunctionDetailsOutput
- Raises:
SessionError – If called before establishing a session via connect().
Example
>>> async with Pctx() as pctx: ... details = await pctx.get_function_details(['Weather.getCurrentWeather']) ... print(details.code) # Detailed TypeScript with parameter info
- async execute(code: str) ExecuteOutput[source]
Execute TypeScript code that calls namespaced functions.
This method runs TypeScript code in a secure Deno sandbox with access to all registered functions (both local tools and MCP server functions).
- Parameters:
code – TypeScript code to execute. Must include an async run() function that serves as the entry point. Functions must be called with their namespace prefix (e.g., ‘Weather.getCurrentWeather()’).
- Returns:
- An object containing execution results with attributes:
result: The value returned from the run() function
logs: Array of console.log() outputs
markdown(): Method to format output as markdown
- Return type:
ExecuteOutput
- Raises:
SessionError – If called before establishing a session via connect().
TimeoutError – If execution exceeds the configured timeout (default 30s).
Notes
Code must define an async function run() as the entry point
Functions MUST be called as ‘Namespace.functionName’
Only functions from list_functions() are available
No access to fetch(), fs, or other standard Node/Deno APIs
Variables don’t persist between execute() calls
Return values are already parsed objects, not JSON strings
Example
>>> async with Pctx() as pctx: ... code = ''' ... async function run() { ... const result = await Weather.getCurrentWeather({ city: "NYC" }); ... console.log("Temperature:", result.temp); ... return { temperature: result.temp }; ... } ... ''' ... output = await pctx.execute(code) ... print(output.markdown()) # Formatted results with logs
- langchain_tools() list[LangchainBaseTool][source]
Expose PCTX code mode tools as langchain tools
- Requires the ‘langchain’ extra to be installed:
pip install pctx[langchain]
- Raises:
ImportError – If langchain is not installed.
- crewai_tools() list[CrewAiBaseTool][source]
Expose PCTX code mode tools as crewai tools
- Requires the ‘crewai’ extra to be installed:
pip install pctx[crewai]
- Raises:
ImportError – If crewai is not installed.
- openai_agents_tools() list[FunctionTool][source]
Expose PCTX code mode tools as OpenAI Agents SDK function tools
- Requires the ‘openai’ extra to be installed:
pip install pctx[openai]
- Returns:
List of function tools compatible with OpenAI Agents SDK
- Raises:
ImportError – If openai is not installed.
- pydantic_ai_tools() list[PydanticAITool][source]
Expose PCTX code mode tools as Pydantic AI tools
- Requires the ‘pydantic-ai’ extra to be installed:
pip install pctx[pydantic-ai]
- Raises:
ImportError – If pydantic-ai is not installed.
Tool Decorator
- pctx_client.tool(name_or_callable: str, *args: Any, namespace: str = 'tools', description: str | None = None) Callable[[Callable], Tool | AsyncTool][source]
- pctx_client.tool(name_or_callable: Callable, *args: Any, namespace: str = 'tools', description: str | None = None) Tool | AsyncTool
Decorator that converts a function into a Tool or AsyncTool instance.
Can be used with or without parameters: - @tool - Uses function name as tool name - @tool(“custom_name”) - Uses custom name for the tool - @tool(namespace=”custom”, description=”…”) - With additional options
- Parameters:
name_or_callable – Either a custom tool name (str) or the function to wrap (Callable)
namespace – The namespace the tool belongs to (default: “tools”)
description – Optional description override (default: uses function docstring)
- Returns:
Either a Tool/AsyncTool instance or a decorator function that creates one
Examples
>>> @tool ... def my_function(x: int) -> int: ... '''Adds one to x''' ... return x + 1
>>> @tool("custom_name", namespace="math") ... def add_two(x: int) -> int: ... return x + 2
Tool Classes
- class pctx_client.Tool(*, name: str, namespace: str, description: str = '', input_schema: Annotated[type[BaseModel] | None, SkipValidation] = None, output_schema: Annotated[Any | None, SkipValidation] = None)[source]
Bases:
BaseTool,ABCSynchronous tool base class
- invoke(**kwargs: Any) Any[source]
Calls the synchronous function with the provided arguments.
- Parameters:
**kwargs – Arguments to pass to the function
- Returns:
The result of the function call
- Raises:
ValueError – If no synchronous function is available
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class pctx_client.AsyncTool(*, name: str, namespace: str, description: str = '', input_schema: Annotated[type[BaseModel] | None, SkipValidation] = None, output_schema: Annotated[Any | None, SkipValidation] = None)[source]
Bases:
BaseTool,ABCAsynchronous tool base class
- async ainvoke(**kwargs: Any) Any[source]
Calls the asynchronous function with the provided arguments.
- Parameters:
**kwargs – Arguments to pass to the function
- Returns:
The result of the function call
- Raises:
ValueError – If no synchronous function is available
- model_config: ClassVar[ConfigDict] = {}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].