# **Graphtomation Documentation**
**⚠️ Disclaimer: This package is still under development. Use it at your own risk.**
---
## Overview
Graphtomation is an AI utility package designed to simplify the development and deployment of AI-powered workflows. By combining Crew and LangGraph with FastAPI, it enables AI engineers to create modular, reusable components and expose them as API endpoints. With tools, agents, tasks, and crews ready for integration, Graphtomation accelerates the process of building and serving complex multi-agent systems.
---
## Installation
Install the required dependencies for Graphtomation using the following command:
```bash
pip install graphtomation
```
---
## Implementation
### Crew
```py
from typing import Type
from fastapi import FastAPI
from crewai.tools import BaseTool
from crewai import Agent, Task, Crew
from pydantic import BaseModel, Field
from langchain_community.tools import DuckDuckGoSearchRun
from graphtomation.crewai import CrewApiRouter, CrewExecutor
app = FastAPI()
class DuckDuckGoSearchInput(BaseModel):
"""Input schema for DuckDuckGoSearchTool."""
query: str = Field(..., description="Search query to look up on DuckDuckGo.")
class DuckDuckGoSearchTool(BaseTool):
name: str = "DuckDuckGoSearch"
description: str = (
"This tool performs web searches using DuckDuckGo and retrieves the top results. "
"Provide a query string to get relevant information."
)
args_schema: Type[BaseModel] = DuckDuckGoSearchInput
def _run(self, query: str) -> str:
"""
Perform a search using the DuckDuckGo API and return the top results.
"""
return DuckDuckGoSearchRun().invoke(query)
ddg_search_tool = DuckDuckGoSearchTool()
researcher = Agent(
role="Web Researcher",
goal="Perform searches to gather relevant information for tasks.",
backstory="An experienced researcher with expertise in online information gathering.",
tools=[ddg_search_tool],
verbose=True,
)
research_task = Task(
description="Search for the latest advancements in AI technology.",
expected_output="A summary of the top 3 advancements in AI technology from recent searches.",
agent=researcher,
)
example_crew = Crew(
agents=[researcher],
tasks=[research_task],
verbose=True,
planning=True,
)
crew_router = CrewApiRouter(
executor=CrewExecutor(
crews=[
{
"name": "example-crew",
"crew": example_crew,
"metadata": {
"description": "An example crew ai implementation",
"version": "1.0.0",
},
}
]
)
)
app.include_router(crew_router.router, prefix="/crew")
```
### Langgraph
```py
import os
from typing import Literal
from fastapi import FastAPI
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
from langchain_community.tools import DuckDuckGoSearchRun
from graphtomation.langgraph import GraphExecutor, GraphApiRouter
from langgraph.graph import END, START, StateGraph, MessagesState
app = FastAPI()
@tool(name_or_callable="search-tool")
def search(query: str):
"""Search the web using this tool"""
return DuckDuckGoSearchRun().invoke(query)
tools = [search]
tool_node = ToolNode(tools)
model = ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY")).bind_tools(tools)
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
def call_model(state: MessagesState):
messages = state["messages"]
response = model.invoke(messages)
return {"messages": [response]}
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
"agent",
should_continue,
)
workflow.add_edge("tools", "agent")
graph_router = GraphApiRouter(
executor=GraphExecutor(
graphs=[
{
"name": "langgraph-chatbot",
"state_graph": workflow,
"kwargs": {
"checkpointer": {
"name": "postgres",
"conn_string": os.getenv("DB_CONN_STRING"),
},
},
}
]
)
)
app.include_router(graph_router.router, prefix="/graphs")
```
Raw data
{
"_id": null,
"home_page": "https://github.com/adimis-toolbox/graphtomation-server/blob/main/packages",
"name": "graphtomation",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": null,
"keywords": "AI utility Crew LangGraph FastAPI API workflows automation",
"author": "Aditya Mishra",
"author_email": "aditya.mishra@adimis.in",
"download_url": "https://files.pythonhosted.org/packages/15/0c/f6b7d8c9185c5790512fcc8cf7b37a6a90b45f5f5f698581e27050805dc7/graphtomation-0.2.0.tar.gz",
"platform": null,
"description": "# **Graphtomation Documentation**\n\n**\u26a0\ufe0f Disclaimer: This package is still under development. Use it at your own risk.**\n\n---\n\n## Overview\n\nGraphtomation is an AI utility package designed to simplify the development and deployment of AI-powered workflows. By combining Crew and LangGraph with FastAPI, it enables AI engineers to create modular, reusable components and expose them as API endpoints. With tools, agents, tasks, and crews ready for integration, Graphtomation accelerates the process of building and serving complex multi-agent systems.\n\n---\n\n## Installation\n\nInstall the required dependencies for Graphtomation using the following command:\n\n```bash\npip install graphtomation\n```\n\n---\n\n## Implementation\n\n### Crew\n\n```py\nfrom typing import Type\nfrom fastapi import FastAPI\nfrom crewai.tools import BaseTool\nfrom crewai import Agent, Task, Crew\nfrom pydantic import BaseModel, Field\nfrom langchain_community.tools import DuckDuckGoSearchRun\nfrom graphtomation.crewai import CrewApiRouter, CrewExecutor\n\n\napp = FastAPI()\n\n\nclass DuckDuckGoSearchInput(BaseModel):\n \"\"\"Input schema for DuckDuckGoSearchTool.\"\"\"\n\n query: str = Field(..., description=\"Search query to look up on DuckDuckGo.\")\n\n\nclass DuckDuckGoSearchTool(BaseTool):\n name: str = \"DuckDuckGoSearch\"\n description: str = (\n \"This tool performs web searches using DuckDuckGo and retrieves the top results. \"\n \"Provide a query string to get relevant information.\"\n )\n args_schema: Type[BaseModel] = DuckDuckGoSearchInput\n\n def _run(self, query: str) -> str:\n \"\"\"\n Perform a search using the DuckDuckGo API and return the top results.\n \"\"\"\n return DuckDuckGoSearchRun().invoke(query)\n\n\nddg_search_tool = DuckDuckGoSearchTool()\n\nresearcher = Agent(\n role=\"Web Researcher\",\n goal=\"Perform searches to gather relevant information for tasks.\",\n backstory=\"An experienced researcher with expertise in online information gathering.\",\n tools=[ddg_search_tool],\n verbose=True,\n)\n\nresearch_task = Task(\n description=\"Search for the latest advancements in AI technology.\",\n expected_output=\"A summary of the top 3 advancements in AI technology from recent searches.\",\n agent=researcher,\n)\n\nexample_crew = Crew(\n agents=[researcher],\n tasks=[research_task],\n verbose=True,\n planning=True,\n)\n\n\ncrew_router = CrewApiRouter(\n executor=CrewExecutor(\n crews=[\n {\n \"name\": \"example-crew\",\n \"crew\": example_crew,\n \"metadata\": {\n \"description\": \"An example crew ai implementation\",\n \"version\": \"1.0.0\",\n },\n }\n ]\n )\n)\n\napp.include_router(crew_router.router, prefix=\"/crew\")\n```\n\n### Langgraph\n\n```py\nimport os\nfrom typing import Literal\nfrom fastapi import FastAPI\nfrom langchain_core.tools import tool\nfrom langchain_openai import ChatOpenAI\nfrom langgraph.prebuilt import ToolNode\nfrom langchain_community.tools import DuckDuckGoSearchRun\nfrom graphtomation.langgraph import GraphExecutor, GraphApiRouter\nfrom langgraph.graph import END, START, StateGraph, MessagesState\n\n\napp = FastAPI()\n\n\n@tool(name_or_callable=\"search-tool\")\ndef search(query: str):\n \"\"\"Search the web using this tool\"\"\"\n return DuckDuckGoSearchRun().invoke(query)\n\n\ntools = [search]\n\ntool_node = ToolNode(tools)\n\nmodel = ChatOpenAI(api_key=os.getenv(\"OPENAI_API_KEY\")).bind_tools(tools)\n\n\ndef should_continue(state: MessagesState) -> Literal[\"tools\", \"__end__\"]:\n messages = state[\"messages\"]\n last_message = messages[-1]\n if last_message.tool_calls:\n return \"tools\"\n return END\n\n\ndef call_model(state: MessagesState):\n messages = state[\"messages\"]\n response = model.invoke(messages)\n return {\"messages\": [response]}\n\n\nworkflow = StateGraph(MessagesState)\n\nworkflow.add_node(\"agent\", call_model)\nworkflow.add_node(\"tools\", tool_node)\n\nworkflow.add_edge(START, \"agent\")\n\nworkflow.add_conditional_edges(\n \"agent\",\n should_continue,\n)\n\nworkflow.add_edge(\"tools\", \"agent\")\n\ngraph_router = GraphApiRouter(\n executor=GraphExecutor(\n graphs=[\n {\n \"name\": \"langgraph-chatbot\",\n \"state_graph\": workflow,\n \"kwargs\": {\n \"checkpointer\": {\n \"name\": \"postgres\",\n \"conn_string\": os.getenv(\"DB_CONN_STRING\"),\n },\n },\n }\n ]\n )\n)\n\napp.include_router(graph_router.router, prefix=\"/graphs\")\n```\n",
"bugtrack_url": null,
"license": null,
"summary": "An AI utility package to build and serve Crew and LangGraph workflows as FastAPI routes, packed with reusable components for AI engineers.",
"version": "0.2.0",
"project_urls": {
"Documentation": "https://github.com/adimis-toolbox/graphtomation-server/blob/main/packages#readme",
"Homepage": "https://github.com/adimis-toolbox/graphtomation-server/blob/main/packages",
"Issue Tracker": "https://github.com/adimis-toolbox/graphtomation-server/issues",
"Source": "https://github.com/adimis-toolbox/graphtomation-server/blob/main/packages"
},
"split_keywords": [
"ai",
"utility",
"crew",
"langgraph",
"fastapi",
"api",
"workflows",
"automation"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "199ebb24137fff7599b1c4124c5e68fd608a2407739aa861ea941e4e4527c998",
"md5": "1700bc07e8b78d348df7ed94d0614825",
"sha256": "2df74ba689b5c2cabdf7a035d5554b08081c15da4d6902a6e856f9b802a7ee35"
},
"downloads": -1,
"filename": "graphtomation-0.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1700bc07e8b78d348df7ed94d0614825",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.7",
"size": 20274,
"upload_time": "2025-01-12T19:26:44",
"upload_time_iso_8601": "2025-01-12T19:26:44.893449Z",
"url": "https://files.pythonhosted.org/packages/19/9e/bb24137fff7599b1c4124c5e68fd608a2407739aa861ea941e4e4527c998/graphtomation-0.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "150cf6b7d8c9185c5790512fcc8cf7b37a6a90b45f5f5f698581e27050805dc7",
"md5": "bea8e9c49fabe563084613b2fc19c9f7",
"sha256": "9bec32c0260b097541bf04f6098350d418627bd96b91bc7c5367c472eb37a4ac"
},
"downloads": -1,
"filename": "graphtomation-0.2.0.tar.gz",
"has_sig": false,
"md5_digest": "bea8e9c49fabe563084613b2fc19c9f7",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 23510,
"upload_time": "2025-01-12T19:26:49",
"upload_time_iso_8601": "2025-01-12T19:26:49.125598Z",
"url": "https://files.pythonhosted.org/packages/15/0c/f6b7d8c9185c5790512fcc8cf7b37a6a90b45f5f5f698581e27050805dc7/graphtomation-0.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-12 19:26:49",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "adimis-toolbox",
"github_project": "graphtomation-server",
"github_not_found": true,
"lcname": "graphtomation"
}
JSON
